fix: Implement reference counting for EventQueue to prevent premature MainQueue closure (#333)

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

Fixes #248 (it became necessary as part of the work)

Author: kabir

.github/workflows/run-tck.yml

@@ -97,6 +97,88 @@ jobs:
           done
           
       - name: Run TCK
+        id: 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: failure()
+        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: failure()
+        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 TCK Log
+        if: failure()
+        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: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: tck-test-log-java-${{ matrix.java-version }}
+          path: tck/target/tck-test.log
+          retention-days: 2
+          if-no-files-found: warn
+      - name: Upload Thread Dump
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: thread-dump-java-${{ matrix.java-version }}
+          path: tck/target/thread-dump.txt
+          retention-days: 2
+          if-no-files-found: warn
+      - name: Upload Heap Dump
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: heap-dump-java-${{ matrix.java-version }}
+          path: tck/target/heap-dump.hprof.gz
+          retention-days: 2
+          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 a core pool size of 5 threads and a maximum pool size of 50 threads, optimized for I/O-bound operations.
+
+You can customize the executor settings in your `application.properties`:
+
+```properties
+# Core thread pool size for the @Internal executor (default: 5)
+a2a.executor.core-pool-size=5
+
+# Maximum thread pool size (default: 50)
+a2a.executor.max-pool-size=50
+
+# Thread keep-alive time in seconds (default: 60)
+a2a.executor.keep-alive-seconds=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.executor.core-pool-size=5
+a2a.executor.max-pool-size=50
+a2a.executor.keep-alive-seconds=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