A Kafka topic is a named, partitioned stream where producers publish records and consumers read them. In an asset-monitoring workflow, asset_location carries location updates from a producer to a consumer. section 5.2
Monitoring asset locations
Create asset_location with one partition, then publish JSON records containing asset_id, latitude, longitude, and timestamp. A Python consumer reads the topic and turns each record into an AssetLocationUpdated event. A mediator can route that event to AssetManager without coupling the Kafka adapter directly to the domain service. section 5.2
A practical topic workflow includes:
- Create and inspect the topic with kafka-topics. section 5.2
- Write records with kafka-console-producer and read them from the beginning with kafka-console-consumer. section 5.2
- Review generated consumer code, especially deserialization, offsets, commits, and architectural coupling, before applying it. section 5.2
Listing 5.13 Docker Compose file to launch Kafka with Zookeeper
version: '2.1'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.3.2
container_name: zookeeper
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_SERVER_ID: 1
ZOOKEEPER_SERVERS: zoo1:2888:3888
kafka:
image: confluentinc/cp-kafka:7.3.2
hostname: kafka
container_name: kafka
ports:
- "9092:9092"
- "29092:29092"
- "9999:9999"
environment:
KAFKA_ADVERTISED_LISTENERS:
INTERNAL://kafka:19092,EXTERNAL://
${DOCKER_HOST_IP:127.0.0.1}:9092,
DOCKER://host.docker.internal:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,
DOCKER:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
KAFKA_BROKER_ID: 1
KAFKA_LOG4J_LOGGERS: "kafka.controller=
INFO,kafka.producer.async
.DefaultEventHandler=INFO,
state.change.logger=INFO"
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_JMX_PORT: 9999
KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1}
KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
depends_on:
- zookeeper
Use the Compose configuration to launch Kafka with Zookeeper for a local development environment. section 5.2
