28 lines
934 B
Python
28 lines
934 B
Python
from kafka import KafkaConsumer
|
|
|
|
topic_name = "keys-topic"
|
|
group_id = "keys-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')
|
|
key = message.key.decode('utf-8')
|
|
print(
|
|
f"Received: {text} with key: {key} from partition: {message.partition}")
|
|
except KeyboardInterrupt:
|
|
print("Stopping consumer...")
|
|
finally:
|
|
consumer.close()
|