Review fixes

Author: kabir

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

@@ -8,7 +8,9 @@
 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;
@@ -237,7 +239,9 @@ public void testKafkaEventReceivedByA2AServer() throws Exception {
             }
         };
 
-        // Create error handler - filter out benign stream closed errors
+        // 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 -> {
             if (!isStreamClosedError(error)) {
                 errorRef.set(error);
@@ -429,9 +433,24 @@ public void testPoisonPillGenerationOnTaskFinalization() throws Exception {
     /**
      * 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) {
-        if (error == null) {
+        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;
         }
 
@@ -449,10 +468,10 @@ private boolean isStreamClosedError(Throwable error) {
             }
         }
 
-        // Check cause recursively
+        // Check cause recursively with cycle detection
         Throwable cause = error.getCause();
-        if (cause != null && cause != error) {
-            return isStreamClosedError(cause);
+        if (cause != null) {
+            return isStreamClosedError(cause, visited);
         }
 
         return false;