learning-kafka
Code Issues

feat: replication example

Browse Source
This commit is contained in:
ruinivist
2026-01-25 04:56:42 +00:00
parent 15c93c71fb
commit 0beb588a70
5 changed files with 178 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -1,5 +1,6 @@
# Kafka data directories
**/kafka-data/
**/kafka-*-data/
**/kafka-logs/
**/zookeeper-data/
06-replication/consumer.py
+24
View File
@@ -0,0 +1,24 @@
from kafka import KafkaConsumer
topic_name = "rep-topic"
group_id = "rep-group"
consumer = KafkaConsumer(
topic_name,
bootstrap_servers=['localhost:9092', 'localhost:9094', 'localhost:9096'],
auto_offset_reset='earliest', # start from the beginning if no offset is committed
enable_auto_commit=True, # read = commit, this is True by default
group_id=group_id
)
try:
for message in consumer:
text = message.value.decode('utf-8')
offset = message.offset
print(
f"Received: {text} from partition: {message.partition} at offset: {offset}")
except KeyboardInterrupt:
print("Stopping consumer...")
finally:
consumer.close()
06-replication/docker-compose.yaml
+107
View File
@@ -0,0 +1,107 @@
services:
kafka-1:
image: apache/kafka:latest
container_name: kafka-1
hostname: kafka-1
ports:
- "9092:9092" # External Client Port
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
# --- NETWORKING CONFIGS ---
# 1. Listeners: Where the process listens inside the container
# CONTROLLER:9093, INTERNAL:29092, EXTERNAL:9092
KAFKA_LISTENERS: CONTROLLER://:9093,INTERNAL://:29092,EXTERNAL://:9092
# 2. Advertised Listeners: What the broker tells others to connect to
# Internal -> within docker network, this network is made by docker-compose
# see `sudo docker network ls`
# External -> for my python script
# these are the two ways you can talk to me
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-1:29092,EXTERNAL://localhost:9092
# 3. Security Map: Define the protocol for the new names
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
# 4. Inter-broker communication MUST use the INTERNAL listener
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL # when brokers talk to each other, use INTERNAL
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER # controller uses CONTROLLER listener
# Quorum voters use the Internal Docker DNS
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
KAFKA_LOG_DIRS: /var/lib/kafka/data
# --- REPLICATION DEFAULTS ---
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
KAFKA_NUM_PARTITIONS: 3
KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_MIN_INSYNC_REPLICAS: 2
volumes:
- ./kafka-1-data:/var/lib/kafka/data
kafka-2:
image: apache/kafka:latest
container_name: kafka-2
hostname: kafka-2
ports:
- "9094:9092" # Host:9094 -> Container:9092
environment:
KAFKA_NODE_ID: 2
KAFKA_PROCESS_ROLES: broker,controller
# Listeners
KAFKA_LISTENERS: CONTROLLER://:9093,INTERNAL://:29092,EXTERNAL://:9092
# Advertised: Internal uses kafka-2, External uses localhost:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-2:29092,EXTERNAL://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
KAFKA_LOG_DIRS: /var/lib/kafka/data
# --- SAME DEFAULTS ---
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
KAFKA_NUM_PARTITIONS: 3
KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_MIN_INSYNC_REPLICAS: 2
volumes:
- ./kafka-2-data:/var/lib/kafka/data
kafka-3:
image: apache/kafka:latest
container_name: kafka-3
hostname: kafka-3
ports:
- "9096:9092" # Host:9096 -> Container:9092
environment:
KAFKA_NODE_ID: 3
KAFKA_PROCESS_ROLES: broker,controller
# Listeners
KAFKA_LISTENERS: CONTROLLER://:9093,INTERNAL://:29092,EXTERNAL://:9092
# Advertised: Internal uses kafka-3, External uses localhost:9096
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-3:29092,EXTERNAL://localhost:9096
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
KAFKA_LOG_DIRS: /var/lib/kafka/data
# --- SAME DEFAULTS ---
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
KAFKA_NUM_PARTITIONS: 3
KAFKA_DEFAULT_REPLICATION_FACTOR: 3
KAFKA_MIN_INSYNC_REPLICAS: 2
volumes:
- ./kafka-3-data:/var/lib/kafka/data
06-replication/produce.py
+34
View File
@@ -0,0 +1,34 @@
# Generates traaffic to be put into the topic
from kafka import KafkaProducer
from kafka.errors import KafkaError
import time
# make a topic
topic_name = "rep-topic" # same as the one made via script
producer = KafkaProducer(
bootstrap_servers=['localhost:9092', 'localhost:9094', 'localhost:9096'],
acks='all', # can be 0 (no ack), 1 (leader ack), all (all replicas ack)
# doing all makes it use the compose setting of min.insync.replicas
retries=5 # number of retries if produce fails
# even if you set in sync replicas in compose, you still need to set acks all here
)
try:
i = 0
while True:
message = f"Message {i}"
meta = producer.send(topic_name, value=message.encode(
'utf-8')).get() # wait for ack
print(
f"Sent: {message} to partition {meta.partition} at offset {meta.offset}")
i += 1
time.sleep(1) # 1 message per second
except KafkaError as ke:
# if two replicas down, not enough replicas error is thrown
print(f"Kafka error occurred: {ke}")
except KeyboardInterrupt:
print("Stopping producer...")
finally:
producer.close()
README.md
+12
View File
@@ -85,3 +85,15 @@ sudo docker exec kafka-ret /opt/kafka/bin/kafka-topics.sh --create --topic ret-t
Kafka moves on from one log file to another based on size or time. These are called "segments". Each segment is a file on disk. Nothing is EVER deleted from the active segment. Only when a segment is "closed" ( rolled over to a new segment ) can it be deleted based on retention policy.
Just retention above won't work, I need to roll it over too.
# 06-replication
I spend a very long time on this one trying to format the kraft storage correctly. Turns out I don't
need to. Modern kafka images auto format it. If you did not get a perm error, you may have missed the
`KAFKA_LOG_DIRS` env var in the docker-compose.yml. I spent too long on this as did not this in my
latest compose. Adding it and giving perms as usual worked perfectly.
Too much trouble on this one but finally done.
If I run the producer and do `sudo docker stop kafka-2` and `sudo docker stop kafka-3`, I get an error on produce as not enough replicas are available. This is because I set `min.insync.replicas=2` in the docker config for default new topics. I also need to set `acks='all'` in the producer too as the default there is `acks=1` which means just leader's ack is fine so even if replicas are down, produce will succeed.
Both settings are needed.