fix: Implement reference counting for EventQueue to prevent premature MainQueue closure

Implements reference counting for EventQueues to fix KafkaReplicationIntegrationTest
and prepare for improvements to the replicated QueueManager. Additional fixes resolve
TCK timeout issues discovered during testing.

## Changes

**EventQueue Reference Counting:**
- MainQueue tracks active ChildQueues with reference counting
- Prevents premature closure when ChildQueues are still consuming events
- Fixes KafkaReplicationIntegrationTest failures
- Prepares infrastructure for replicated QueueManager improvements

**Fix ForkJoinPool Saturation (TCK fixes):**
- Inject @internal Executor (15 threads) into all transport handlers
- Changed CompletableFuture.runAsync() to use injected executor instead of ForkJoinPool.commonPool()
- Prevents streaming subscription timeouts under concurrent load on CI (3 threads)
- Affects: RestHandler, GrpcHandler, JSONRPCHandler

**Improved Queue Lifecycle:**
- Removed awaitQueuePollerStart() to eliminate thread blocking bottleneck
- EventConsumer now manages queue closing on terminal events
- Background cleanup to avoid blocking request threads
- Fixed race condition when clients disconnect during streaming
- Better terminal event detection in ResultAggregator

**Tests & Documentation:**
- Updated all transport handler tests to pass executor parameter
- Added executor configuration section to README.md

TCK tests now pass reliably on CI.

Author: kabir

.github/workflows/run-tck.yml

@@ -97,6 +97,87 @@ jobs:
           done
           
       - name: Run TCK
+        timeout-minutes: 5
         run: |
           ./run_tck.py --sut-url ${{ env.SUT_JSONRPC_URL }} --category all --transports jsonrpc,grpc,rest --compliance-report report.json
         working-directory: tck/a2a-tck
+      - name: Capture Thread Dump
+        if: always()
+        run: |
+          # Find the actual Quarkus JVM (child of Maven process), not the Maven parent
+          # Look for the dev.jar process which is the actual application
+          QUARKUS_PID=$(pgrep -f "a2a-tck-server-dev.jar" || echo "")
+          if [ -n "$QUARKUS_PID" ]; then
+            echo "📊 Capturing thread dump for Quarkus JVM PID $QUARKUS_PID"
+            jstack $QUARKUS_PID > tck/target/thread-dump.txt || echo "Failed to capture thread dump"
+            if [ -f tck/target/thread-dump.txt ]; then
+              echo "✅ Thread dump captured ($(wc -l < tck/target/thread-dump.txt) lines)"
+            fi
+          else
+            echo "⚠️ No Quarkus JVM process found for thread dump"
+            echo "Available Java processes:"
+            ps aux | grep java || true
+          fi
+      - name: Capture Heap Dump
+        if: always()
+        run: |
+          # Find the actual Quarkus JVM (child of Maven process), not the Maven parent
+          QUARKUS_PID=$(pgrep -f "a2a-tck-server-dev.jar" || echo "")
+          if [ -n "$QUARKUS_PID" ]; then
+            echo "📊 Capturing heap dump for Quarkus JVM PID $QUARKUS_PID"
+            jmap -dump:live,format=b,file=tck/target/heap-dump.hprof $QUARKUS_PID || echo "Failed to capture heap dump"
+            if [ -f tck/target/heap-dump.hprof ]; then
+              SIZE=$(du -h tck/target/heap-dump.hprof | cut -f1)
+              echo "✅ Heap dump captured ($SIZE)"
+              # Compress to reduce artifact size
+              gzip tck/target/heap-dump.hprof
+              COMPRESSED_SIZE=$(du -h tck/target/heap-dump.hprof.gz | cut -f1)
+              echo "✅ Compressed heap dump ($COMPRESSED_SIZE)"
+            fi
+          else
+            echo "⚠️ No Quarkus JVM process found for heap dump"
+            echo "Available Java processes:"
+            ps aux | grep java || true
+          fi
+      - name: Stop Quarkus Server
+        if: always()
+        run: |
+          # Find and kill the Quarkus process to ensure logs are flushed
+          pkill -f "quarkus:dev" || true
+          sleep 2
+      - name: Verify and Upload TCK Log
+        if: always()
+        run: |
+          echo "Checking for log file..."
+          if [ -f tck/target/tck-test.log ]; then
+            echo "✅ Log file exists ($(wc -l < tck/target/tck-test.log) lines)"
+            ls -lh tck/target/tck-test.log
+          else
+            echo "❌ Log file not found at tck/target/tck-test.log"
+            echo "Contents of tck/target/:"
+            ls -la tck/target/ || echo "tck/target/ does not exist"
+          fi
+      - name: Upload TCK Log
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: tck-test-log-java-${{ matrix.java-version }}
+          path: tck/target/tck-test.log
+          retention-days: 7
+          if-no-files-found: warn
+      - name: Upload Thread Dump
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: thread-dump-java-${{ matrix.java-version }}
+          path: tck/target/thread-dump.txt
+          retention-days: 1
+          if-no-files-found: warn
+      - name: Upload Heap Dump
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: heap-dump-java-${{ matrix.java-version }}
+          path: tck/target/heap-dump.hprof.gz
+          retention-days: 1
+          if-no-files-found: warn

.serena/cache/java/document_symbols_cache_v23-06-25.pkl

@@ -232,6 +232,38 @@ public class WeatherAgentExecutorProducer {
 }
 ```
 
+### 4. Configure Executor Settings (Optional)
+
+The A2A Java SDK uses a dedicated executor for handling asynchronous operations like streaming subscriptions. By default, this executor is configured with 15 threads optimized for I/O-bound operations.
+
+You can customize the executor settings in your `application.properties`:
+
+```properties
+# Thread pool size for the @Internal executor (default: 15)
+a2a.internal.executor.core-pool-size=15
+
+# Maximum thread pool size (default: 30)
+a2a.internal.executor.max-pool-size=30
+
+# Thread keep-alive time in seconds (default: 60)
+a2a.internal.executor.keep-alive-time=60
+```
+
+**Why this matters:**
+- **Streaming Performance**: The executor handles streaming subscriptions. Too few threads can cause timeouts under concurrent load.
+- **Resource Management**: The dedicated executor prevents streaming operations from competing with the ForkJoinPool used by other async tasks.
+- **Concurrency**: In production environments with high concurrent streaming requests, increase the pool sizes accordingly.
+
+**Default Configuration:**
+```properties
+# These are the defaults - no need to set unless you want different values
+a2a.internal.executor.core-pool-size=15
+a2a.internal.executor.max-pool-size=30
+a2a.internal.executor.keep-alive-time=60
+```
+
+**Note:** The reference server implementations automatically configure this executor. If you're creating a custom server integration, ensure you provide an `@Internal Executor` bean for optimal streaming performance.
+
 ## A2A Client
 
 The A2A Java SDK provides a Java client implementation of the [Agent2Agent (A2A) Protocol](https://google-a2a.github.io/A2A), allowing communication with A2A servers. The Java client implementation supports the following transports:

README.md

@@ -94,6 +94,11 @@ public EventQueue.EventQueueBuilder getEventQueueBuilder(String taskId) {
                 .hook(new ReplicationHook(taskId));
     }
 
+    @Override
+    public int getActiveChildQueueCount(String taskId) {
+        return delegate.getActiveChildQueueCount(taskId);
+    }
+
     private class ReplicatingEventQueueFactory implements EventQueueFactory {
         @Override
         public EventQueue.EventQueueBuilder builder(String taskId) {

extras/queue-manager-replicated/core/src/main/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManager.java

@@ -6,6 +6,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -17,6 +18,7 @@
 
 import io.a2a.server.events.EventQueue;
 import io.a2a.server.events.EventQueueClosedException;
+import io.a2a.server.events.EventQueueTestHelper;
 import io.a2a.spec.Event;
 import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.TaskState;
@@ -147,8 +149,11 @@ void testBasicQueueManagerFunctionality() throws InterruptedException {
         EventQueue queue = queueManager.createOrTap(taskId);
         assertNotNull(queue);
 
+        // createOrTap now returns ChildQueue, get returns MainQueue
         EventQueue retrievedQueue = queueManager.get(taskId);
-        assertEquals(queue, retrievedQueue);
+        assertNotNull(retrievedQueue);
+        // queue should be a ChildQueue (cannot be tapped)
+        assertThrows(IllegalStateException.class, () -> EventQueueTestHelper.tapQueue(queue));
 
         EventQueue tappedQueue = queueManager.tap(taskId);
         assertNotNull(tappedQueue);

extras/queue-manager-replicated/core/src/test/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java

@@ -0,0 +1,10 @@
+package io.a2a.server.events;
+
+/**
+ * Utils to access package private methods in the io.a2a.server.events package
+ */
+public class EventQueueTestHelper {
+    public static EventQueue tapQueue(EventQueue queue) {
+        return queue.tap();
+    }
+}

extras/queue-manager-replicated/core/src/test/java/io/a2a/server/events/EventQueueTestHelper.java

@@ -3,8 +3,11 @@
 import jakarta.enterprise.inject.Instance;
 import jakarta.inject.Inject;
 
+import java.util.concurrent.Executor;
+
 import io.a2a.server.PublicAgentCard;
 import io.a2a.server.requesthandlers.RequestHandler;
+import io.a2a.server.util.async.Internal;
 import io.a2a.spec.AgentCard;
 import io.a2a.transport.grpc.handler.CallContextFactory;
 import io.a2a.transport.grpc.handler.GrpcHandler;
@@ -20,14 +23,17 @@ public class QuarkusGrpcHandler extends GrpcHandler {
     private final AgentCard agentCard;
     private final RequestHandler requestHandler;
     private final Instance<CallContextFactory> callContextFactoryInstance;
+    private final Executor executor;
 
     @Inject
     public QuarkusGrpcHandler(@PublicAgentCard AgentCard agentCard,
                               RequestHandler requestHandler,
-                              Instance<CallContextFactory> callContextFactoryInstance) {
+                              Instance<CallContextFactory> callContextFactoryInstance,
+                              @Internal Executor executor) {
         this.agentCard = agentCard;
         this.requestHandler = requestHandler;
         this.callContextFactoryInstance = callContextFactoryInstance;
+        this.executor = executor;
     }
 
     @Override
@@ -44,4 +50,9 @@ protected AgentCard getAgentCard() {
     protected CallContextFactory getCallContextFactory() {
         return callContextFactoryInstance.isUnsatisfied() ? null : callContextFactoryInstance.get();
     }
+
+    @Override
+    protected Executor getExecutor() {
+        return executor;
+    }
 }

reference/grpc/src/main/java/io/a2a/server/grpc/quarkus/QuarkusGrpcHandler.java

@@ -105,6 +105,14 @@ public Response getStreamingSubscribedCount() {
         return Response.ok(String.valueOf(streamingSubscribedCount.get()), TEXT_PLAIN).build();
     }
 
+    @GET
+    @Path("/queue/childCount/{taskId}")
+    @Produces(TEXT_PLAIN)
+    public Response getChildQueueCount(@PathParam("taskId") String taskId) {
+        int count = testUtilsBean.getChildQueueCount(taskId);
+        return Response.ok(String.valueOf(count), TEXT_PLAIN).build();
+    }
+
     @DELETE
     @Path("/task/{taskId}/config/{configId}")
     public Response deleteTaskPushNotificationConfig(@PathParam("taskId") String taskId, @PathParam("configId") String configId) {

reference/grpc/src/test/java/io/a2a/server/grpc/quarkus/A2ATestResource.java

@@ -139,6 +139,14 @@ public void getStreamingSubscribedCount(RoutingContext rc) {
                 .end(String.valueOf(streamingSubscribedCount.get()));
     }
 
+    @Route(path = "/test/queue/childCount/:taskId", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+    public void getChildQueueCount(@Param String taskId, RoutingContext rc) {
+        int count = testUtilsBean.getChildQueueCount(taskId);
+        rc.response()
+                .setStatusCode(200)
+                .end(String.valueOf(count));
+    }
+
     @Route(path = "/test/task/:taskId/config/:configId", methods = {Route.HttpMethod.DELETE}, type = Route.HandlerType.BLOCKING)
     public void deleteTaskPushNotificationConfig(@Param String taskId, @Param String configId, RoutingContext rc) {
         try {

reference/jsonrpc/src/test/java/io/a2a/server/apps/quarkus/A2ATestRoutes.java

@@ -1 +1 @@
-quarkus.arc.selected-alternatives=io.a2a.server.apps.common.TestHttpClient
+quarkus.arc.selected-alternatives=io.a2a.server.apps.common.TestHttpClient