feat: retry mechanism for some kafka tests

Author: s.kazemi

features/steps/kafka_adapter_steps.py

@@ -1,4 +1,5 @@
 # features/steps/kafka_steps.py
+import time
 from behave import given, then, when
 from confluent_kafka import TopicPartition
 from features.test_helpers import get_current_scenario_context
@@ -57,6 +58,21 @@ def get_kafka_consumer_adapter(context, topic_name, group_id):
         setattr(scenario_context, consumer_key, consumer)
     return getattr(scenario_context, consumer_key)
 
+def wait_for_topic_condition(adapter, condition_func, topic_name, max_retries=5, initial_delay=0.5):
+    """Helper function to wait for a topic condition with retries."""
+    delay = initial_delay
+    for attempt in range(max_retries):
+        try:
+            topics = adapter.list_topics(timeout=2).topics
+            if condition_func(topic_name, topics):
+                return True
+        except Exception as e:
+            if attempt == max_retries - 1:
+                raise e
+        if attempt < max_retries - 1:
+            time.sleep(delay)
+            delay *= 1.5
+    return False
 
 # Given steps
 @given("a configured Kafka admin adapter")
@@ -174,37 +190,28 @@ def step_delete_topic(context, topic_name):
 @then('the topic "{topic_name}" should exist')
 def step_topic_should_exist(context, topic_name):
     adapter = get_kafka_admin_adapter(context)
-    try:
-        topics = adapter.list_topics(timeout=1).topics
-        assert topic_name in topics, f"Topic '{topic_name}' does not exist"
+    if wait_for_topic_condition(adapter, lambda name, topics: name in topics, topic_name):
         context.logger.info(f"Verified topic '{topic_name}' exists")
-    except Exception as e:
-        context.logger.exception(f"Failed to verify topic existence: {str(e)}")
-        raise
+    else:
+        raise AssertionError(f"Topic '{topic_name}' does not exist after retries")
 
 
 @then('the topic "{topic_name}" should not exist')
 def step_topic_should_not_exist(context, topic_name):
     adapter = get_kafka_admin_adapter(context)
-    try:
-        topics = adapter.list_topics(timeout=1).topics
-        assert topic_name not in topics, f"Topic '{topic_name}' still exists"
+    if wait_for_topic_condition(adapter, lambda name, topics: name not in topics, topic_name):
         context.logger.info(f"Verified topic '{topic_name}' does not exist")
-    except Exception as e:
-        context.logger.exception(f"Failed to verify topic non-existence: {str(e)}")
-        raise
+    else:
+        raise AssertionError(f"Topic '{topic_name}' still exists after retries")
 
 
 @then('the topic list should include "{topic_name}"')
 def step_topic_list_includes(context, topic_name):
     adapter = get_kafka_admin_adapter(context)
-    try:
-        topics = adapter.list_topics(timeout=1).topics
-        assert topic_name in topics, f"Topic '{topic_name}' not in topic list"
+    if wait_for_topic_condition(adapter, lambda name, topics: name in topics, topic_name):
         context.logger.info(f"Verified '{topic_name}' in topic list")
-    except Exception as e:
-        context.logger.exception(f"Failed to verify topic list: {str(e)}")
-        raise
+    else:
+        raise AssertionError(f"Topic '{topic_name}' not in topic list after retries")
 
 
 @then('the consumer should receive message "{expected_message}" from topic "{topic_name}"')