Made Kafka topic configurable via KAFKA_TOPIC environment variable for producer and consumer.

- Updated `producer.py` and `consumer.py` to support configurable topic names.
- Refactored `Dockerfiles` to include `KAFKA_TOPIC` environment variable setup.
- Modified `docker-compose.yml` to pass `KAFKA_TOPIC` to all relevant services.

Author: starichkov

consumer.Dockerfile

@@ -10,14 +10,9 @@ RUN pip install --no-cache-dir -r requirements.txt
 COPY logger.py .
 COPY consumer.py .
 
-# Set environment variable for Kafka connection
+# Set environment variables for Kafka connection
 ENV KAFKA_BOOTSTRAP_SERVERS=kafka:9092
-
-# Modify the consumer.py to use the environment variable
-RUN sed -i "s/'bootstrap_servers': .*/'bootstrap_servers': os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092'),/" consumer.py
-
-# Add import os at the top of the file
-RUN sed -i '1s/^/import os\n/' consumer.py
+ENV KAFKA_TOPIC=test-topic
 
 # Command to run the consumer
 # The entrypoint allows passing additional arguments to the consumer

consumer.py

@@ -18,7 +18,8 @@
 It can handle both JSON and plain text messages, providing a polyglot consumer
 that's useful in environments where different systems produce data in different formats.
 
-The consumer connects to a local Kafka broker, subscribes to 'test-topic',
+The consumer connects to a Kafka broker (configurable via KAFKA_BOOTSTRAP_SERVERS environment variable),
+subscribes to a topic (configurable via KAFKA_TOPIC environment variable, defaults to 'test-topic'),
 and processes incoming messages until interrupted with Ctrl+C.
 
 Usage:
@@ -51,9 +52,10 @@ def try_parse_json(value: bytes):
 
 def main():
     bootstrap_servers = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
+    topic = os.environ.get("KAFKA_TOPIC", "test-topic")
 
     # Initialize the Kafka consumer with configuration
-    # - 'test-topic': The Kafka topic to subscribe to
+    # - topic: The Kafka topic to subscribe to (from environment variable)
     # - bootstrap_servers: Connection string for the Kafka broker
     # - auto_offset_reset='earliest': Start reading from the beginning of the topic if no committed offset exists
     # - enable_auto_commit=True: Automatically commit offsets
@@ -67,11 +69,9 @@ def main():
         consumer_args['group_id'] = args.group_id
 
     if args.test_mode:
-        consumer_timeout_ms = 3000  # pragma: no cover
-    else:
-        consumer_timeout_ms = None
+        consumer_args['consumer_timeout_ms'] = 3000  # pragma: no cover
 
-    consumer = KafkaConsumer('test-topic', consumer_timeout_ms=consumer_timeout_ms, **consumer_args)
+    consumer = KafkaConsumer(topic, **consumer_args)
 
     logger = get_logger("consumer")
 

docker-compose.yml

@@ -46,6 +46,7 @@ services:
         condition: service_completed_successfully
     environment:
       KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+      KAFKA_TOPIC: test-topic
     restart: on-failure
 
   consumer:
@@ -60,6 +61,7 @@ services:
         condition: service_completed_successfully
     environment:
       KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+      KAFKA_TOPIC: test-topic
     restart: on-failure
     # You can pass arguments to the consumer here
     # command: ["--group-id", "demo-group"]

producer.Dockerfile

@@ -10,13 +10,8 @@ RUN pip install --no-cache-dir -r requirements.txt
 COPY logger.py .
 COPY producer.py .
 
-# Set environment variable for Kafka connection
+# Set environment variables for Kafka connection
 ENV KAFKA_BOOTSTRAP_SERVERS=kafka:9092
-
-# Modify the producer.py to use the environment variable
-RUN sed -i 's/bootstrap_servers=.*/bootstrap_servers=os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"),/' producer.py
-
-# Add import os at the top of the file
-RUN sed -i '1s/^/import os\n/' producer.py
+ENV KAFKA_TOPIC=test-topic
 
 CMD ["python", "-u", "producer.py"]

producer.py

@@ -15,8 +15,9 @@
 Apache Kafka Producer Demo
 
 This script demonstrates how to produce JSON messages to an Apache Kafka topic.
-It creates a series of simple JSON messages and sends them to 'test-topic' on
-a local Kafka broker.
+It creates a series of simple JSON messages and sends them to a topic
+(configurable via KAFKA_TOPIC environment variable, defaults to 'test-topic')
+on a Kafka broker (configurable via KAFKA_BOOTSTRAP_SERVERS environment variable).
 
 The producer is configured to automatically serialize Python dictionaries to JSON
 before sending them to Kafka.
@@ -30,6 +31,7 @@
 
 def main():
     bootstrap_servers = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
+    topic = os.environ.get("KAFKA_TOPIC", "test-topic")
 
     # Initialize the Kafka producer with configuration
     # - bootstrap_servers: Connection string for the Kafka broker
@@ -48,8 +50,8 @@ def main():
         # Create a simple message with an ID and text
         message = {"id": i, "event_type": event_type, "text": f"Note event {i} of type {event_type}"}
 
-        # Send the message to 'test-topic'
-        producer.send('test-topic', key=event_type, value=message)
+        # Send the message to the topic from the environment variable
+        producer.send(topic, key=event_type, value=message)
 
         # Print confirmation and wait 1 second between messages
         logger.info(f"Sent: key={event_type} | value={message}")