from kafka import KafkaConsumer

topic_name = "ret-topic"
group_id = "ret-group"

consumer = KafkaConsumer(
    topic_name,
    bootstrap_servers='localhost:9092',  # advertised listener
    auto_offset_reset='earliest',  # start from the beginning if no offset is committed
    enable_auto_commit=True,  # read = commit, this is True by default
    # who am I in the consumer group, without it you'll always read from beginning
    # as kafka does not know if you have read before as you had not name for yourself
    group_id=group_id

)

try:
    for message in consumer:
        # message is of type ConsumerRecord and data is in value which is bytes
        text = message.value.decode('utf-8')
        offset = message.offset
        print(
            f"Received: {text} from partition: {message.partition} at offset: {offset}")
except KeyboardInterrupt:
    print("Stopping consumer...")
finally:
    consumer.close()