[Python] Consuming Kafka message in Python

I am creating a micro service app that uses Kafka. There are two services: one built with Golang(message producer) and another one built with Python(Kafka consumer). And they are running via Docker. The problem is that Golang publishes messages without any problem, but the consumer is not working. I can verify messages are published on the Kafka ui but am not able to see consumer created.

version: '3.8'
services:
fraud-service:
build: ../fraud-service
depends_on:
kafka:
condition: service_healthy
postgres:
condition: service_healthy
volumes:
- ../fraud-service:/app
environment:
- KAFKA_BROKER=kafka:9092
- KAFKA_TOPIC=transaction.created
- DB_URL=dbname=transferdb user=transferuser password=secret host=postgres

kafka:
image: confluentinc/cp-kafka:latest
ports:
- "9092:9092"
environment:
healthcheck:
test: ["CMD", "bash", "-c", "</dev/tcp/localhost/9092"]
interval: 10s
timeout: 5s
retries: 10
volumes:
- kafka-data:/var/lib/kafka/data
kafka-ui:
image: provectuslabs/kafka-ui:latest
ports:
- "8080:8080"
environment:
- KAFKA_CLUSTERS_0_NAME=local
- KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
depends_on:
- kafka


Here is the snippet of fraud-service(Python) code

logging.info(KAFKA_BROKER)
conn = get_db_connection()
consumer = KafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=[KAFKA_BROKER],
auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='fraud-consumer-group',
value_deserializer=lambda x: json.loads(x.decode('utf-8')),
)
for message in consumer:
logging.info(f" Message received (partition {message.partition}, offset {message.offset})")


screenshot of Kafka UI I have verified that messages are coming into the Topic, but the Python service is not subscribing to it. I tried to remove the deserializer and tried the polling approach, but it still doesn't work. This is all the log I can see in the service.

2025-03-15 20:58:28,391 [INFO] Starting Fraud Detection Service on topic: transaction.created
2025-03-15 20:58:28,391 [INFO] kafka:9092
2025-03-15 20:58:28,397 [INFO] <BrokerConnection client_id=kafka-python-2.0.6, node_id=bootstrap-0 host=kafka:9092 <connecting> [IPv4 ('172.18.0.4', 9092)]>: connecting to kafka:9092 [('172.18.0.4', 9092) IPv4]
2025-03-15 20:58:28,397 [INFO] Probing node bootstrap-0 broker version
2025-03-15 20:58:28,397 [INFO] <BrokerConnection client_id=kafka-python-2.0.6, node_id=bootstrap-0 host=kafka:9092 <connecting> [IPv4 ('172.18.0.4', 9092)]>: Connection complete.
2025-03-15 20:58:28,499 [INFO] Broker version identified as 2.6.0
2025-03-15 20:58:28,499 [INFO] Set configuration api_version=(2, 6, 0) to skip auto check_version requests on startup
2025-03-15 20:58:28,500 [INFO] Updating subscribed topics to: ('transaction.created',)
2025-03-15 20:58:28,703 [INFO] <BrokerConnection client_id=kafka-python-2.0.6, node_id=1 host=kafka:9092 <connecting> [IPv4 ('172.18.0.4', 9092)]>: connecting to kafka:9092 [('172.18.0.4', 9092) IPv4]
2025-03-15 20:58:28,703 [INFO] <BrokerConnection client_id=kafka-python-2.0.6, node_id=1 host=kafka:9092 <connecting> [IPv4 ('172.18.0.4', 9092)]>: Connection complete.
2025-03-15 20:58:28,703 [INFO] <BrokerConnection client_id=kafka-python-2.0.6, node_id=bootstrap-0 host=kafka:9092 <connected> [IPv4 ('172.18.0.4', 9092)]>: Closing connection.

Continue reading...