-
Notifications
You must be signed in to change notification settings - Fork 0
/
consumer.py
36 lines (30 loc) · 1.25 KB
/
consumer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from confluent_kafka import Consumer, KafkaError, KafkaException
# Adjusted consumer configuration
config = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'new-test-group', # Consider using a new group ID for a fresh start
'auto.offset.reset': 'earliest' # Ensure we start from the beginning of the topic
}
consumer = Consumer(**config)
consumer.subscribe(['topic_python_medium_demo'])
def consume_messages():
try:
while True:
msg = consumer.poll(timeout=1.0) # Increase poll timeout if needed
if msg is None:
continue # No message, normal behavior
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event, normal behavior but can adjust if not desired
print(f'End of partition reached {msg.topic()}/{msg.partition()}')
else:
print(f'Error: {msg.error()}')
else:
# Message is successfully received
print(f"Received message: {msg.value().decode('utf-8')}")
except KeyboardInterrupt:
print("Stopping consumer...")
finally:
consumer.close()
if __name__ == "__main__":
consume_messages()