from pykafka import KafkaClient
from pykafka.common import OffsetType

host = 'quickstart.cloudera'
port = '9092'
hosts = host + ':' + port
hosts = hosts
client = KafkaClient(hosts=hosts)

print('topics found:')
for topic in client.topics:
    print(topic.decode('utf-8'))

#topic = client.topics['test'.encode('utf-8')]
topic = client.topics[b'test']
print('partitions', topic.partitions)
print('latest_available_offsets', topic.latest_available_offsets())

consumer = topic.get_simple_consumer(consumer_group=b"mygroup",
                                     auto_offset_reset=OffsetType.LATEST,
                                     reset_offset_on_start=True)
print('simple consumer offsets', consumer.fetch_offsets())
print('consuming messages')
for message in consumer:
    if message is not None:
        print(message.offset, message.value.decode('utf-8'))
print('waiting around')
