Clean up event queue logging a bit

Author: kabir

core/src/main/java/io/a2a/server/events/EventConsumer.java

@@ -16,8 +16,6 @@
 public class EventConsumer {
     private final EventQueue queue;
     private Throwable error;
-    private final Executor executor = Executors.newCachedThreadPool();
-
 
     private static final String ERROR_MSG = "Agent did not return any response";
     private static final int NO_WAIT = -1;

core/src/main/java/io/a2a/server/events/EventQueue.java

@@ -6,6 +6,7 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.a2a.util.TempLoggerWrapper;
 import org.slf4j.Logger;
@@ -35,7 +36,9 @@ public static EventQueue create() {
         return new MainQueue();
     }
 
-    abstract CountDownLatch getPollingStartedLatch();
+    public abstract void awaitQueuePollerStart() throws InterruptedException ;
+
+    abstract void signalQueuePollerStarted();
 
     public void enqueueEvent(Event event) {
         if (closed) {
@@ -71,13 +74,12 @@ public Event dequeueEvent(int waitMilliSeconds) throws EventQueueClosedException
                 }
                 return event;
             } catch (InterruptedException e) {
-                log.debug("Interrupted {}", this);
+                log.debug("Interrupted dequeue (waiting) {}", this);
                 Thread.currentThread().interrupt();
                 return null;
             }
         } finally {
-            log.debug("Signalling that queue polling started {}", this);
-            getPollingStartedLatch().countDown();
+            signalQueuePollerStarted();
         }
     }
 
@@ -103,7 +105,8 @@ public void close() {
 
     static class MainQueue extends EventQueue {
         private final List<ChildQueue> children = new CopyOnWriteArrayList<>();
-        private CountDownLatch pollingStartedLatch = new CountDownLatch(1);
+        private final CountDownLatch pollingStartedLatch = new CountDownLatch(1);
+        private final AtomicBoolean pollingStarted = new AtomicBoolean(false);
 
         EventQueue tap() {
             ChildQueue child = new ChildQueue(this);
@@ -116,11 +119,22 @@ public void enqueueEvent(Event event) {
             children.forEach(eq -> eq.internalEnqueueEvent(event));
         }
 
-        CountDownLatch getPollingStartedLatch() {
-            return pollingStartedLatch;
+        @Override
+        public void awaitQueuePollerStart() throws InterruptedException {
+            log.debug("Waiting for queue poller to start om {}", this);
+            pollingStartedLatch.await(10, TimeUnit.SECONDS);
+            log.debug("Queue poller started on {}", this);
         }
 
-
+        @Override
+        void signalQueuePollerStarted() {
+            if (pollingStarted.get()) {
+                return;
+            }
+            log.debug("Signalling that queue polling started {}", this);
+            pollingStartedLatch.countDown();
+            pollingStarted.set(true);
+          }
 
         @Override
         public void close() {
@@ -151,8 +165,13 @@ EventQueue tap() {
         }
 
         @Override
-        CountDownLatch getPollingStartedLatch() {
-            return parent.getPollingStartedLatch();
+        public void awaitQueuePollerStart() throws InterruptedException {
+            parent.awaitQueuePollerStart();
+        }
+
+        @Override
+        void signalQueuePollerStarted() {
+            parent.signalQueuePollerStarted();
         }
     }
 }

core/src/main/java/io/a2a/server/events/InMemoryQueueManager.java

@@ -7,10 +7,12 @@
 
 import jakarta.enterprise.context.ApplicationScoped;
 
+import io.a2a.util.TempLoggerWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 @ApplicationScoped
 public class InMemoryQueueManager implements QueueManager {
-
-
     private final Map<String, EventQueue> queues = Collections.synchronizedMap(new HashMap<>());
 
     @Override
@@ -63,7 +65,7 @@ public EventQueue createOrTap(String taskId) {
     }
 
     @Override
-    public void signalPollingStarted(EventQueue eventQueue) throws InterruptedException {
-        eventQueue.getPollingStartedLatch().await(10, TimeUnit.SECONDS);
+    public void awaitQueuePollerStart(EventQueue eventQueue) throws InterruptedException {
+        eventQueue.awaitQueuePollerStart();
     }
 }

core/src/main/java/io/a2a/server/events/QueueManager.java

@@ -11,5 +11,5 @@ public interface QueueManager {
 
     EventQueue createOrTap(String taskId);
 
-    void signalPollingStarted(EventQueue eventQueue) throws InterruptedException;
+    void awaitQueuePollerStart(EventQueue eventQueue) throws InterruptedException;
 }

core/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java

@@ -355,7 +355,7 @@ private EnhancedRunnable registerAndExecuteAgentAsync(String taskId, RequestCont
             public void run() {
                 agentExecutor.execute(requestContext, queue);
                 try {
-                    queueManager.signalPollingStarted(queue);
+                    queueManager.awaitQueuePollerStart(queue);
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                 }

core/src/main/java/io/a2a/server/requesthandlers/JSONRPCHandler.java

@@ -1,6 +1,5 @@
 package io.a2a.server.requesthandlers;
 
-import static io.a2a.util.AsyncUtils.convertingProcessor;
 import static io.a2a.util.AsyncUtils.createTubeConfig;
 
 import java.util.concurrent.Flow;