diff --git a/.gitignore b/.gitignore index 9125522..f34e41d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Kafka data directories **/kafka-data/ +**/kafka-*-data/ **/kafka-logs/ **/zookeeper-data/ diff --git a/06-replication/consumer.py b/06-replication/consumer.py new file mode 100644 index 0000000..374692f --- /dev/null +++ b/06-replication/consumer.py @@ -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() diff --git a/06-replication/docker-compose.yaml b/06-replication/docker-compose.yaml new file mode 100644 index 0000000..ff519f6 --- /dev/null +++ b/06-replication/docker-compose.yaml @@ -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 diff --git a/06-replication/produce.py b/06-replication/produce.py new file mode 100644 index 0000000..cf511bb --- /dev/null +++ b/06-replication/produce.py @@ -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() diff --git a/README.md b/README.md index a99c77f..29dd2f7 100644 --- a/README.md +++ b/README.md @@ -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.