2

I wrote a program to consume kafka events. it has a daemon which i want to terminate after 10s .

def kafkaConsumer():
consumer = KafkaConsumer(sys.argv[1],group_id='test-consumer-group',bootstrap_servers=sys.argv[2].split(','))
schema_path=sys.argv[3]
schema = avro.schema.parse(open(schema_path).read())

for msg in consumer:
    value = bytearray(msg.value)
    bytes_reader = io.BytesIO(value[9:])
    decoder = avro.io.BinaryDecoder(bytes_reader)
    reader = avro.io.DatumReader(schema)
    try:
        event = reader.read(decoder)
    except:
        pass
    eventInJsonFormat=json.dumps(event)
    print(eventInJsonFormat)

if __name__ == '__main__':
run_thread = Thread(target=kafkaConsumer())
run_thread.daemon = True
run_thread.start()
time.sleep(10)

Please ignore indentation.
But this program is not terminating after 10s. Want to know what am i missing here ?

vaibhav.g
  • 729
  • 1
  • 9
  • 28
  • 1
    have you tried terminating the thread, not just sleeping?http://stackoverflow.com/questions/16262132/how-terminate-python-thread-without-checking-flag-continuously – beoliver Dec 03 '15 at 08:49

1 Answers1

0

Add a 10s timeout to your consumer:

consumer = KafkaConsumer(sys.argv[1],
consumer_timeout_ms=10000
group_id='test-consumer-group',
bootstrap_servers=sys.argv[2].split(','),
)
Josl
  • 43
  • 1
  • 3