feat: rebalance b/w paritions example

02-python-conn/docker-compose.yaml

@@ -3,7 +3,7 @@ version: "3"
 services:
     kafka:
         image: apache/kafka:latest
-        container_name: kafka-python-conn
+        container_name: kafka-python
         ports:
             - "127.0.0.1:9092:9092" # localhost interface,host:container
         environment:

03-rebalance/consumer.py

@@ -0,0 +1,25 @@
+from kafka import KafkaConsumer
+
+topic_name = "rebal-topic"
+group_id = "rebal-group"
+
+consumer = KafkaConsumer(
+    topic_name,
+    bootstrap_servers='localhost:9092',  # advertised listener
+    auto_offset_reset='earliest',  # start from the beginning if no offset is committed
+    enable_auto_commit=True,  # read = commit, this is True by default
+    # who am I in the consumer group, without it you'll always read from beginning
+    # as kafka does not know if you have read before as you had not name for yourself
+    group_id=group_id
+
+)
+
+try:
+    for message in consumer:
+        # message is of type ConsumerRecord and data is in value which is bytes
+        text = message.value.decode('utf-8')
+        print(f"Received: {text} from partition: {message.partition}")
+except KeyboardInterrupt:
+    print("Stopping consumer...")
+finally:
+    consumer.close()

03-rebalance/docker-compose.yaml

@@ -0,0 +1,34 @@
+version: "3"
+
+services:
+    kafka:
+        image: apache/kafka:latest
+        container_name: kafka-rebalance
+        ports:
+            - "127.0.0.1:9092:9092" # localhost interface,host:container
+        environment:
+            # who am I in the cluster
+            KAFKA_NODE_ID: 1 # node identifier in cluster
+            KAFKA_PROCESS_ROLES: broker,controller # Combined broker + controller (KRaft mode, no Zookeeper)
+            # how to vote in the cluster, service name at line 4
+            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 # Controller nodes for quorum (node_id@host:port)
+
+            # Listener configuration
+            # what interfaces and ports on the container layer ( not host )
+            # PLAINTEXT and CONTROLLER are listener names, can be anything as long as
+            # referenced the correct ones after
+            KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 # Ports Kafka listens on internally
+            # outside container layer ( host )
+            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 # Address advertised to clients
+            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT # encryption settings for listeners
+            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER # listener name for controller communication
+
+            KAFKA_LOG_DIRS: /var/lib/kafka/data # Directory for message data storage
+
+            # when running locally you NEED this else consumer can't read via group
+            # as by default it wants 3 replicas for these internal topics and errors out
+            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 # replication factor for offsets topic
+            KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 # replication factor for transaction log
+            KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 # how many replicas must acknowledge a write
+        volumes:
+            - ./kafka-data:/var/lib/kafka/data # Persist data across container restarts

03-rebalance/produce.py

@@ -0,0 +1,29 @@
+# Generates traaffic to be put into the topic
+
+from kafka import KafkaProducer
+import time
+
+# make a topic
+topic_name = "rebal-topic"  # same as the one made via script
+
+# producer instance
+producer = KafkaProducer(
+    bootstrap_servers='localhost:9092')  # advertised listener
+
+try:
+    i = 0
+    while True:
+        # i NEED to encode to bytes, when I did not, failed with this assert
+        # assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))
+        message = f"Message {i}"
+        # or I can define a value_serializer
+        producer.send(topic_name, value=message.encode(
+            'utf-8')).get()  # wait for ack
+        # on the producer to do the encoding automatically
+        print(f"Sent: {message}")
+        i += 1
+        time.sleep(1)  # 1 message per second
+except KeyboardInterrupt:
+    print("Stopping producer...")
+finally:
+    producer.close()

README.md

@@ -1,6 +1,6 @@
 learning kafka
 
-# Setup
+# 01-broken-setup ( not actually broken )
 
 Take a look at 01-broken-setup. Ideally when you setup, you need to have a meta.properties file in the kafka-data ( /var/lib/kafka/data ) directory. This file is created when Kafka starts for the first time.
 
@@ -16,3 +16,37 @@ version=1
 If this file is missing, the docker image will make one. But in case you run multiple brokers, all need to have the same cluster.id. So they can talk to each other. This is known as generate/format Kraft storage.
 
 Neede to run: `sudo chown -R 1000:1000 kafka-data/` to give the kafka user ( uid 1000 ) permission to write to the volume mounted directory. First 1000 is user id, second 1000 is group id.
+
+# 02-python-conn
+
+simple example, 1 producer, 1 consumer, no partitions or replication
+
+# 03-rebalance
+
+5 partitions
+
+`kafka-rebalance` is the container name
+topic is the `rebal-topic`
+
+```
+sudo docker exec kafka-rebalance /opt/kafka/bin/kafka-topics.sh --create --topic rebal-topic --partitions 5 --bootstrap-server localhost:9092
+```
+
+Run one producer and one consumer. The consumer will be assigned all 5 partitions.
+The messages will be out of order for the partitions.
+
+Then in another terminal, run another consumer in the same group, then another similarly. The paritions will be rebalanced between the consumers automatically. If I kill one consumer, the partitions will be rebalanced again to the remaining consumers.
+
+To see what the group assignments are, run:
+
+```
+sudo docker exec -it kafka-rebalance /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group rebal-group
+```
+
+This will show what partitions are assigned to what consumers in the group.
+
+If you run another producer, messages will be twice as fast and all "just works".
+
+If you create more consumers than paritions, the extra consumers will be idle.
+
+There is no scaling down of partitions, to decrease partitions, you need to create a new topic with less partitions and migrate data over.