[Python] Kafka Producing and Consuming in Python does not work. Kafka in Docker, and...

Trying to get Kafka running in Docker on a Macbook.

• Starting Kafka works.
• Creating a Topic from inside the Docker Container works (with the provided Bash script)
• Creating a Topic from a Python script on the Host works. (See Python Script below)

But what does not work:

• Producing and Consuming messages from a Python script on the Host. Just nothing happens, and nothing to see in the Docker Logs

Libraries used


The following Python libraries are used in the Python Scripts:

• kafka-python-ng==2.2.3
• pybind11==2.13.6

Docker Compose file


The docker-compose.yml is taken from Apache/Kafka on Docker Hub. Only change is the volumes which is mapped to the Host, and the 9092 port is mapped to the Host port 9092.

services:
broker:
image: apache/kafka:latest
container_name: kafka
volumes:
- ./data:/tmp/kraft-combined-logs:rw
ports:
- '9092:9092'
environment:
KAFKA_NODE_ID: 1
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLERLAINTEXT,PLAINTEXTLAINTEXT,PLAINTEXT_HOSTLAINTEXT'
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT_HOST://localhost:9092,PLAINTEXT://broker:19092'
KAFKA_PROCESS_ROLES: 'broker,controller'
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
KAFKA_LISTENERS: 'CONTROLLER://:29093,PLAINTEXT_HOST://:9092,PLAINTEXT://:19092'
KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
CLUSTER_ID: 'sdfgvkdmpritgnvsyr56'
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1
KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1
KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'


Kafka does start and runs without any errors.

Creating a Topic


from kafka import KafkaAdminClient
from kafka.admin import NewTopic, ConfigResource, ConfigResourceType
from kafka.errors import TopicAlreadyExistsError


if __name__ == '__main__':
admin_client = KafkaAdminClient(bootstrap_servers='localhost:9092')

# Create a Topic
topic_list = []
new_topic = NewTopic(name="bankbranch", num_partitions=2, replication_factor=1)
topic_list.append(new_topic)

try:
admin_client.create_topics(new_topics=topic_list)
except TopicAlreadyExistsError:
print('*** Topic already exists: bankbranch')


See the topics


Inside the Docker Container

9f30328dd248:/opt/kafka/bin$ ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic
Created topic test-topic.

9f30328dd248:/opt/kafka/bin$ ./kafka-topics.sh --bootstrap-server localhost:9092 --list
bankbranch
test-topic


So, the create-topic script works.

kafka-producer.py


from kafka import KafkaProducer
import json


if __name__ == '__main__':
producer = KafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

producer.send("bankbranch", {'atmid': 1, 'transid': 100})
producer.send("bankbranch", {'atmid': 2, 'transid': 101})

kafka-consumer.py


from kafka import KafkaConsumer

if __name__ == '__main__':
consumer = KafkaConsumer('bankbranch',
bootstrap_servers='localhost:9092',
auto_offset_reset='latest',
)
for msg in consumer:
print(msg.value.decode("utf-8"))


Testing Producer en Consumer


I start both on the Host and no response. No errors, just nothing.

Producer returns to the prompt, and Consumer just keeps waiting.

Continue reading...