diff --git a/04-keys/consumer.py b/04-keys/consumer.py new file mode 100644 index 0000000..419a179 --- /dev/null +++ b/04-keys/consumer.py @@ -0,0 +1,27 @@ +from kafka import KafkaConsumer + +topic_name = "keys-topic" +group_id = "keys-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') + key = message.key.decode('utf-8') + print( + f"Received: {text} with key: {key} from partition: {message.partition}") +except KeyboardInterrupt: + print("Stopping consumer...") +finally: + consumer.close() diff --git a/04-keys/docker-compose.yaml b/04-keys/docker-compose.yaml new file mode 100644 index 0000000..76cdf9e --- /dev/null +++ b/04-keys/docker-compose.yaml @@ -0,0 +1,34 @@ +version: "3" + +services: + kafka: + image: apache/kafka:latest + container_name: kafka-keys + 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 diff --git a/04-keys/produce.py b/04-keys/produce.py new file mode 100644 index 0000000..3488d3c --- /dev/null +++ b/04-keys/produce.py @@ -0,0 +1,33 @@ +# Generates traaffic to be put into the topic + +from kafka import KafkaProducer +import time + +# make a topic +topic_name = "keys-topic" # same as the one made via script + +# producer instance +producer = KafkaProducer( + bootstrap_servers='localhost:9092') # advertised listener + +try: + i = 0 + users = ['Alice', 'Bob', 'Charlie'] + + while True: + for user in users: + message = f"Message {i} from {user}" + + # add the user bytes as key + meta = producer.send(topic_name, key=user.encode('utf-8'), value=message.encode( + # wait for ack, synchronous so NEVER use in prod + 'utf-8')).get() + print( + f"Sent: {message} with key: {user} to partition: {meta.partition}") + + i += 1 + time.sleep(1) # 1 message per second +except KeyboardInterrupt: + print("Stopping producer...") +finally: + producer.close() diff --git a/README.md b/README.md index 687badf..9e8da5b 100644 --- a/README.md +++ b/README.md @@ -52,3 +52,22 @@ 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. # 04-keys + +If I want to have all messages for a certain user to always go in the same partition, I can use keys. +Example: if messages for the user MUST be processed in order. + +Change the container name and the topic + +``` +sudo chown -R 1000:1000 kafka-data + +sudo docker exec kafka-keys /opt/kafka/bin/kafka-topics.sh --create --topic keys-topic --partitions 2 --bootstrap-server localhost:9092 +``` + +If you forget to run the above command to create the topic, the topic will be created automatically when the producer sends the first message. But it will have only 1 partition by default. + +DELETE topic: + +``` +docker exec -it kafka-keys /opt/kafka/bin/kafka-topics.sh --delete --topic keys-topic --bootstrap-server localhost:9092 +```