import sys
from kafka import KafkaConsumer
from redlock import Redlock

if len(sys.argv) < 4:
    print('Invoke the script in the form "python kafka-consumer.py kafka_connection_string topic traces_path"')
    exit()

kafka_connection = sys.argv[1]
topic = sys.argv[2]
traces_path = sys.argv[3]

if not traces_path[-1] == '/':
    traces_path += "/"

# To consume messages
consumer = KafkaConsumer(topic,
                         group_id='traces_cache',
                         bootstrap_servers=[kafka_connection])
consumer.commit()
dlm = Redlock([{"host": "localhost", "port": 6379, "db": 0}, ])
for message in consumer:
    key = message.key.decode('utf-8')
    lock = False
    while not lock:
        lock = dlm.lock(key, 5000)

    with open(traces_path + key, 'ab') as f:
        f.write(message.value)

    dlm.unlock(lock)

    consumer.task_done(message)
    consumer.commit()

d8a3e720933414a006a786576da709cb81ed9588
<mask token>

<mask token>
if len(sys.argv) < 4:
    print(
        'Invoke the script in the form "python kafka-consumer.py kafka_connection_string topic traces_path"'
        )
    exit()
<mask token>
if not traces_path[-1] == '/':
    traces_path += '/'
<mask token>
consumer.commit()
<mask token>
for message in consumer:
    key = message.key.decode('utf-8')
    lock = False
    while not lock:
        lock = dlm.lock(key, 5000)
    with open(traces_path + key, 'ab') as f:
        f.write(message.value)
    dlm.unlock(lock)
    consumer.task_done(message)
    consumer.commit()

<mask token>
if len(sys.argv) < 4:
    print(
        'Invoke the script in the form "python kafka-consumer.py kafka_connection_string topic traces_path"'
        )
    exit()
kafka_connection = sys.argv[1]
topic = sys.argv[2]
traces_path = sys.argv[3]
if not traces_path[-1] == '/':
    traces_path += '/'
consumer = KafkaConsumer(topic, group_id='traces_cache', bootstrap_servers=
    [kafka_connection])
consumer.commit()
dlm = Redlock([{'host': 'localhost', 'port': 6379, 'db': 0}])
for message in consumer:
    key = message.key.decode('utf-8')
    lock = False
    while not lock:
        lock = dlm.lock(key, 5000)
    with open(traces_path + key, 'ab') as f:
        f.write(message.value)
    dlm.unlock(lock)
    consumer.task_done(message)
    consumer.commit()

import sys
from kafka import KafkaConsumer
from redlock import Redlock
if len(sys.argv) < 4:
    print(
        'Invoke the script in the form "python kafka-consumer.py kafka_connection_string topic traces_path"'
        )
    exit()
kafka_connection = sys.argv[1]
topic = sys.argv[2]
traces_path = sys.argv[3]
if not traces_path[-1] == '/':
    traces_path += '/'
consumer = KafkaConsumer(topic, group_id='traces_cache', bootstrap_servers=
    [kafka_connection])
consumer.commit()
dlm = Redlock([{'host': 'localhost', 'port': 6379, 'db': 0}])
for message in consumer:
    key = message.key.decode('utf-8')
    lock = False
    while not lock:
        lock = dlm.lock(key, 5000)
    with open(traces_path + key, 'ab') as f:
        f.write(message.value)
    dlm.unlock(lock)
    consumer.task_done(message)
    consumer.commit()

import sys
from kafka import KafkaConsumer
from redlock import Redlock

if len(sys.argv) < 4:
    print('Invoke the script in the form "python kafka-consumer.py kafka_connection_string topic traces_path"')
    exit()

kafka_connection = sys.argv[1]
topic = sys.argv[2]
traces_path = sys.argv[3]

if not traces_path[-1] == '/':
    traces_path += "/"

# To consume messages
consumer = KafkaConsumer(topic,
                         group_id='traces_cache',
                         bootstrap_servers=[kafka_connection])
consumer.commit()
dlm = Redlock([{"host": "localhost", "port": 6379, "db": 0}, ])
for message in consumer:
    key = message.key.decode('utf-8')
    lock = False
    while not lock:
        lock = dlm.lock(key, 5000)

    with open(traces_path + key, 'ab') as f:
        f.write(message.value)

    dlm.unlock(lock)

    consumer.task_done(message)
    consumer.commit()
