fix: kafka connection issues

Author: s.kazemi

features/steps/kafka_adapter_steps.py

@@ -11,9 +11,14 @@ def get_kafka_admin_adapter(context):
     """Get or initialize the Kafka admin adapter."""
     scenario_context = get_current_scenario_context(context)
     if not hasattr(scenario_context, "admin_adapter") or scenario_context.admin_adapter is None:
-        test_config = scenario_context.get("test_config")
-        context.logger.info("Initializing Kafka admin adapter")
-        scenario_context.admin_adapter = KafkaAdminAdapter(test_config.KAFKA)
+        # Get the updated configuration from the running container
+        test_containers = scenario_context.get("test_containers")
+        kafka_container = test_containers.get_container("kafka")
+
+        # Use the configuration from the running container
+        kafka_config = kafka_container.config
+
+        scenario_context.admin_adapter = KafkaAdminAdapter(kafka_config)
     return scenario_context.admin_adapter
 
 
@@ -24,9 +29,14 @@ def get_kafka_producer_adapter(context, topic_name):
         not hasattr(scenario_context, f"producer_{topic_name}")
         or getattr(scenario_context, f"producer_{topic_name}") is None
     ):
-        test_config = scenario_context.get("test_config")
-        context.logger.info(f"Initializing Kafka producer for topic: {topic_name}")
-        producer = KafkaProducerAdapter(topic_name, kafka_configs=test_config.KAFKA)
+        # Get the updated configuration from the running container
+        test_containers = scenario_context.get("test_containers")
+        kafka_container = test_containers.get_container("kafka")
+
+        # Use the configuration from the running container
+        kafka_config = kafka_container.config
+
+        producer = KafkaProducerAdapter(topic_name, kafka_configs=kafka_config)
         setattr(scenario_context, f"producer_{topic_name}", producer)
     return getattr(scenario_context, f"producer_{topic_name}")
 
@@ -36,9 +46,14 @@ def get_kafka_consumer_adapter(context, topic_name, group_id):
     scenario_context = get_current_scenario_context(context)
     consumer_key = f"consumer_{topic_name}_{group_id}"
     if not hasattr(scenario_context, consumer_key) or getattr(scenario_context, consumer_key) is None:
-        test_config = scenario_context.get("test_config")
-        context.logger.info(f"Initializing Kafka consumer for topic: {topic_name}, group: {group_id}")
-        consumer = KafkaConsumerAdapter(group_id=group_id, topic_list=[topic_name], kafka_configs=test_config.KAFKA)
+        # Get the updated configuration from the running container
+        test_containers = scenario_context.get("test_containers")
+        kafka_container = test_containers.get_container("kafka")
+
+        # Use the configuration from the running container
+        kafka_config = kafka_container.config
+
+        consumer = KafkaConsumerAdapter(group_id=group_id, topic_list=[topic_name], kafka_configs=kafka_config)
         setattr(scenario_context, consumer_key, consumer)
     return getattr(scenario_context, consumer_key)
 

features/test_containers.py

@@ -356,7 +356,7 @@ def start(self) -> KafkaContainer:
         # Set container properties from running container
         self.host = self._container.get_container_host_ip()
         self.bootstrap_servers = self._container.get_bootstrap_server()
-        self.config.BROKERS_LIST = self.bootstrap_servers
+        self.config.BROKERS_LIST = [self.bootstrap_servers]
 
         logger.info("Kafka container started on %s:%s", self.host, self.port)
         logger.info("Bootstrap servers: %s", self.bootstrap_servers)