diff --git a/commands.md b/commands.md index 56d87695895716bf064207352c229466735592ed..e1012420ce43df78ba5aff542aa776562e73c58c 100644 --- a/commands.md +++ b/commands.md @@ -1,5 +1,5 @@ -## SETTING UP PYTHON ENVIRONMENT +## SETTING UP PYTHON ENVIRONMENT (2 different containers) - Activate venv @@ -9,19 +9,25 @@ [] pip install -r requirements.txt +- Run Kafka Producer / Consumer + +[] python kafka_consumer.py +[] python kafka_producer.py + ## SETTING UP KAFKA SERVER - Download Kafka server [] wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz [] tar -xzf kafka_2.13-3.9.0.tgz -[] cd kafka_2.13-3.9.0 - Setup env variables -[] KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh dragons-1)" +[] cd kafka_2.13-3.9.0 +[] KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)" [] bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties - Start server -[] bin/kafka-server-start.sh config/kraft/reconfig-server.properties \ No newline at end of file +[] bin/kafka-server-start.sh config/kraft/reconfig-server.properties + diff --git a/kafka_consumer.py b/kafka_consumer.py index 5b96aa14d1b5c7ee4a38d23d0e402e41c632f7c6..a546e4c9b26d6047be54277a0c85f909e3cef2b5 100644 --- a/kafka_consumer.py +++ b/kafka_consumer.py @@ -19,7 +19,7 @@ try: if msg is None: continue if msg.error(): - if msg.error().code() == KafkaError._PARTITION_EOF: + if msg.error().code() == KafkaException._PARTITION_EOF: # End of partition event print(f"{msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}") elif msg.error():