kafka_python.py

import confluent_kafka
#import json
#import msgpack

c = confluent_kafka.Consumer({'bootstrap.servers': 'kafka', 'group.id': 'PyCharm_consumer',
              'default.topic.config': {'auto.offset.reset': 'latest'}})

c.subscribe(['ping'])
running = True
i = 0
while running:
    msg = c.poll()
    if not msg.error():
        i = i + 1
        j = msg.value()
        #j = json.loads(msgpack.unpackb(msg.value()))
        #j = json.loads(msg.value())
        print str(i) + ": " + str(j)
    elif msg.error().code() != confluent_kafka.KafkaError._PARTITION_EOF:
        print(msg.error())
        running = False
c.close()