fix: Filter benign stream errors in KafkaReplicationIntegrationTest a…

.github/workflows/build-and-test.yml

@@ -27,3 +27,23 @@ jobs:
         cache: maven
     - name: Build with Maven and run tests
       run: mvn -B package --file pom.xml -fae
+    - name: Upload Test Reports
+      if: failure()
+      uses: actions/upload-artifact@v4
+      with:
+        name: surefire-reports-java-${{ matrix.java-version }}
+        path: |
+          **/target/surefire-reports/
+          **/target/failsafe-reports/
+        retention-days: 7
+        if-no-files-found: warn
+    - name: Upload Build Logs
+      if: failure()
+      uses: actions/upload-artifact@v4
+      with:
+        name: build-logs-java-${{ matrix.java-version }}
+        path: |
+          **/target/*.log
+          **/target/quarkus.log
+        retention-days: 3
+        if-no-files-found: ignore

.github/workflows/run-tck.yml

@@ -100,85 +100,71 @@ jobs:
         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
+          ./run_tck.py --sut-url ${{ env.SUT_JSONRPC_URL }} --category all --transports jsonrpc,grpc,rest --compliance-report report.json 2>&1 | tee tck-output.log
         working-directory: tck/a2a-tck
-      - name: Capture Thread Dump
+      - name: Capture Diagnostics on Failure
         if: failure()
         run: |
+          echo "=== Capturing diagnostic information ==="
+
+          # Create diagnostics directory
+          mkdir -p tck/target/diagnostics
+
+          # Capture process list
+          echo "📋 Capturing process list..."
+          ps auxww > tck/target/diagnostics/processes.txt
+
           # 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)"
+            jstack $QUARKUS_PID > tck/target/diagnostics/thread-dump.txt || echo "Failed to capture thread dump"
+            if [ -f tck/target/diagnostics/thread-dump.txt ]; then
+              echo "✅ Thread dump captured ($(wc -l < tck/target/diagnostics/thread-dump.txt) lines)"
             fi
           else
             echo "⚠️ No Quarkus JVM process found for thread dump"
             echo "Available Java processes:"
-            ps aux | grep java || true
+            ps aux | grep java | tee -a tck/target/diagnostics/processes.txt || 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
+
+          # Capture Quarkus application logs (if available)
+          echo "📝 Checking for Quarkus logs..."
+          if [ -f tck/target/quarkus.log ]; then
+            cp tck/target/quarkus.log tck/target/diagnostics/
+            echo "✅ Copied quarkus.log ($(wc -l < tck/target/quarkus.log) lines)"
+          fi
+
+          # Copy TCK server logs
+          if [ -f tck/target/tck-test.log ]; then
+            cp tck/target/tck-test.log tck/target/diagnostics/
+            echo "✅ Copied tck-test.log ($(wc -l < tck/target/tck-test.log) lines)"
           fi
+
+          echo ""
+          echo "=== Diagnostic capture complete ==="
       - 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
+      - name: Upload TCK Diagnostics
         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
+          name: tck-diagnostics-java-${{ matrix.java-version }}
+          path: |
+            tck/target/diagnostics/
+            tck/a2a-tck/tck-output.log
+          retention-days: 7
           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()
+      - name: Upload TCK Compliance Report
+        if: always()
         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
+          name: tck-compliance-report-java-${{ matrix.java-version }}
+          path: tck/a2a-tck/report.json
+          retention-days: 14
+          if-no-files-found: ignore

extras/queue-manager-replicated/tests-single-instance/src/test/java/io/a2a/extras/queuemanager/replicated/tests/KafkaReplicationIntegrationTest.java

@@ -7,7 +7,10 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.IOException;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -236,9 +239,13 @@ public void testKafkaEventReceivedByA2AServer() throws Exception {
             }
         };
 
-        // Create error handler
+        // Create error handler - filter out benign stream closed errors.
+        // HTTP/2 streams are cancelled during normal cleanup when subscriptions end,
+        // which is expected behavior and not an actual error condition.
         Consumer<Throwable> errorHandler = error -> {
-            errorRef.set(error);
+            if (!isStreamClosedError(error)) {
+                errorRef.set(error);
+            }
             resubscribeLatch.countDown();
         };
 
@@ -423,4 +430,51 @@ public void testPoisonPillGenerationOnTaskFinalization() throws Exception {
         assertEquals(taskId, closedEvent.getTaskId(), "QueueClosedEvent task ID should match");
     }
 
+    /**
+     * Checks if an error is a benign stream closed/cancelled error that should be ignored.
+     * HTTP/2 streams can be cancelled during normal cleanup, which is not an actual error.
+     *
+     * @param error the throwable to check (may be null)
+     * @return true if this is a benign stream closure error that should be ignored
+     */
+    private boolean isStreamClosedError(Throwable error) {
+        return isStreamClosedError(error, new HashSet<>());
+    }
+
+    /**
+     * Internal recursive implementation with cycle detection to prevent infinite recursion.
+     *
+     * @param error the throwable to check
+     * @param visited set of already-visited throwables to detect cycles
+     * @return true if this is a benign stream closure error
+     */
+    private boolean isStreamClosedError(Throwable error, Set<Throwable> visited) {
+        if (error == null || !visited.add(error)) {
+            // Null or already visited (cycle detected)
+            return false;
+        }
+
+        // Check for IOException which includes stream cancellation
+        if (error instanceof IOException) {
+            String message = error.getMessage();
+            if (message != null) {
+                // Filter out normal stream closure/cancellation errors
+                if (message.contains("Stream closed") ||
+                    message.contains("Stream") && message.contains("cancelled") ||
+                    message.contains("EOF reached") ||
+                    message.contains("CANCEL")) {
+                    return true;
+                }
+            }
+        }
+
+        // Check cause recursively with cycle detection
+        Throwable cause = error.getCause();
+        if (cause != null) {
+            return isStreamClosedError(cause, visited);
+        }
+
+        return false;
+    }
+
 }