fix: Improve Kubernetes Example reliability on CI (#435)

- exit early once we have the two required pods
- increased count until we expect 2 pods

Fixes #396 🦕

Author: kabir

.github/workflows/cloud-deployment-example.yml

@@ -58,7 +58,8 @@ jobs:
           mvn test-compile exec:java \
             -Dexec.mainClass="io.a2a.examples.cloud.A2ACloudExampleClient" \
             -Dexec.classpathScope=test \
-            -Dagent.url=http://localhost:8080
+            -Dagent.url=http://localhost:8080 \
+            -Dci.mode=true
 
       - name: Show diagnostics on failure
         if: failure()

examples/cloud-deployment/server/src/test/java/io/a2a/examples/cloud/A2ACloudExampleClient.java

@@ -1,5 +1,15 @@
 package io.a2a.examples.cloud;
 
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import io.a2a.A2A;
 import io.a2a.client.Client;
 import io.a2a.client.ClientEvent;
@@ -9,26 +19,13 @@
 import io.a2a.client.config.ClientConfig;
 import io.a2a.client.transport.jsonrpc.JSONRPCTransport;
 import io.a2a.client.transport.jsonrpc.JSONRPCTransportConfigBuilder;
-import io.a2a.spec.A2AClientError;
-import io.a2a.spec.A2AClientException;
 import io.a2a.spec.AgentCard;
 import io.a2a.spec.Message;
 import io.a2a.spec.Part;
-import io.a2a.spec.Task;
 import io.a2a.spec.TaskArtifactUpdateEvent;
 import io.a2a.spec.TaskIdParams;
 import io.a2a.spec.TextPart;
 
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
 /**
  * Test client demonstrating multi-pod A2A agent deployment with modernized message protocol.
  * <p>
@@ -50,8 +47,11 @@
 public class A2ACloudExampleClient {
 
     private static final String AGENT_URL = System.getProperty("agent.url", "http://localhost:8080");
-    private static final int PROCESS_MESSAGE_COUNT = 8;  // Number of "process" messages to send
-    private static final int MESSAGE_INTERVAL_MS = 1500;
+    private static final int PROCESS_MESSAGE_COUNT = Integer.parseInt(System.getProperty("process.message.count", "8"));  // Number of "process" messages to send
+    private static final int MESSAGE_INTERVAL_MS = Integer.parseInt(System.getProperty("message.interval.ms", "1500"));
+    private static final boolean CI_MODE = Boolean.parseBoolean(System.getProperty("ci.mode", "false"));  // Early exit when 2 pods observed
+    private static final int MAX_MESSAGES_UNTIL_TWO_PODS = Integer.parseInt(System.getProperty("max.messages.until.two.pods", "20"));  // Max messages before giving up
+    private static final int MIN_PODS_TO_OBSERVE = 2;
 
     // Test state
     private final Map<String, Integer> observedPods = Collections.synchronizedMap(new HashMap<>());
@@ -253,11 +253,19 @@ private void handleSubscriptionError(Throwable error) {
 
     private void sendProcessMessages() throws InterruptedException {
         System.out.println();
-        System.out.println("Step 3: Sending " + PROCESS_MESSAGE_COUNT + " 'process' messages (interval: " + MESSAGE_INTERVAL_MS + "ms)...");
+        if (CI_MODE) {
+            System.out.println("Step 3: Sending 'process' messages until 2 pods observed (CI mode, max: " + MAX_MESSAGES_UNTIL_TWO_PODS + ", interval: " + MESSAGE_INTERVAL_MS + "ms)...");
+        } else {
+            System.out.println("Step 3: Sending " + PROCESS_MESSAGE_COUNT + " 'process' messages (interval: " + MESSAGE_INTERVAL_MS + "ms)...");
+        }
         System.out.println("--------------------------------------------");
 
-        for (int i = 1; i <= PROCESS_MESSAGE_COUNT; i++) {
-            final int messageNum = i;
+        int messageCount = 0;
+        int maxMessages = CI_MODE ? MAX_MESSAGES_UNTIL_TWO_PODS : PROCESS_MESSAGE_COUNT;
+        
+        while (messageCount < maxMessages) {
+            messageCount++;
+            final int messageNum = messageCount;
 
             // Create a new client for each request to force new HTTP connection
             Client freshClient = Client.builder(streamingClient.getAgentCard())
@@ -282,8 +290,15 @@ private void sendProcessMessages() throws InterruptedException {
                 });
 
                 Thread.sleep(MESSAGE_INTERVAL_MS);
+                
+                // In CI mode, check if we've observed 2 pods and can exit early
+                if (CI_MODE && observedPods.size() >= 2) {
+                    System.out.println();
+                    System.out.println("✓ CI mode: Successfully observed 2 pods after " + messageNum + " messages. Stopping early.");
+                    break;
+                }
             } catch (Exception e) {
-                System.err.println("✗ Failed to send process message " + i + ": " + e.getMessage());
+                System.err.println("✗ Failed to send process message " + messageNum + ": " + e.getMessage());
                 testFailed.set(true);
             }
         }
@@ -345,8 +360,8 @@ private void printResults() {
         if (testFailed.get()) {
             System.out.println("✗ TEST FAILED - Errors occurred during execution");
             System.exit(1);
-        } else if (observedPods.size() < 2) {
-            System.out.println("✗ TEST FAILED - Expected at least 2 different pods, but only saw: " + observedPods.size());
+        } else if (observedPods.size() < MIN_PODS_TO_OBSERVE) {
+            System.out.printf("✗ TEST FAILED - Expected at least %d different pods, but only saw: %d\n", MIN_PODS_TO_OBSERVE, observedPods.size());
             System.out.println("  This suggests load balancing is not working correctly.");
             System.exit(1);
         } else {