Cherry pick branch 'genexuslabs:fix/kafka' into beta

Author: claudiamurialdo

.github/workflows/Kafka-Integration-Tests.yml

@@ -54,6 +54,17 @@ jobs:
         dotnet sln $SolutionFile add dotnet/src/extensions/kafka/src/Core.Messaging.csproj
         dotnet sln $SolutionFile add dotnet/src/extensions/kafka/test/DotNetCoreKafkaTest/DotNetCoreKafkaTest.csproj 
       
+    - name: Wait for Kafka to be ready
+      run: |
+        for i in {1..30}; do
+          if nc -z localhost 9092; then
+            echo "Kafka is ready!";
+            break;
+          fi;
+          echo "Waiting for Kafka...";
+          sleep 5;
+        done
+        
     - name: Test Kafka
       env:
         KAFKA_TEST_ENABLED: "true"

dotnet/src/extensions/kafka/test/DotNetKafkaTest/KafkaTest.cs

@@ -20,19 +20,22 @@ public KafkaIntegrationTests()
 		public void Test_Producer_And_Consumer_With_Kafka()
 		{
 			GXMessaging producer = new GXMessaging();
-			producer.Configuration = "{'bootstrap.servers': 'localhost:9092', 'default.topic.config': {'message.timeout.ms': 10000}}";
+			producer.Configuration = "{'bootstrap.servers': 'kafka:9092', 'default.topic.config': {'message.timeout.ms': 30000}}";
 
 			GXMessaging consumer = new GXMessaging();
-			consumer.Configuration = "{'bootstrap.servers': 'localhost:9092', 'group.id': 'test-group', 'auto.offset.reset': 'earliest', 'session.timeout.ms': 10000}";
+			consumer.Configuration = "{'bootstrap.servers': 'kafka:9092', 'group.id': 'test-group', 'auto.offset.reset': 'earliest', 'session.timeout.ms': 30000}";
 
 			string key = "msgkey_1";
 			string message = "Integration test message";
-
+			Console.WriteLine($"Producing message with key: {key}");
 			//Produce a message
 			producer.ProduceAsync(_topic, key, message);
+			Console.WriteLine($"Message produced: {message}");
 			producer.Finish(10000);
 
+			Console.WriteLine($"Consuming message from topic: {_topic}");
 			var messages = consumer.Consume(_topic, 10000);
+			Console.WriteLine($"Messages consumed: {messages.Count}");
 
 			//Verify the message was consumed
 			Assert.Single(messages);