[LOGMGR-343] Check the thread id and name of a record when entering t…

…he AsyncHandler and do not add it to the queue if there is a match.

https://issues.redhat.com/browse/LOGMGR-343
Signed-off-by: James R. Perkins <jperkins@redhat.com>

src/main/java/org/jboss/logmanager/handlers/AsyncHandler.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.Consumer;
 import java.util.logging.Handler;
 
 import org.jboss.logmanager.ExtHandler;
@@ -34,10 +35,13 @@
  * some degree of latency.
  */
 public class AsyncHandler extends ExtHandler {
+    public static final Consumer<ExtLogRecord> DISCARDING_REENTRY_HANDLER = record -> {
+    };
 
     private final BlockingQueue<ExtLogRecord> recordQueue;
     private final int queueLength;
     private final Thread thread;
+    private final Consumer<ExtLogRecord> reentryHandler;
     private volatile OverflowAction overflowAction = OverflowAction.BLOCK;
 
     @SuppressWarnings("unused")
@@ -55,13 +59,7 @@ public class AsyncHandler extends ExtHandler {
      * @param threadFactory the thread factory to use to construct the handler thread
      */
     public AsyncHandler(final int queueLength, final ThreadFactory threadFactory) {
-        recordQueue = new ArrayBlockingQueue<ExtLogRecord>(queueLength);
-        thread = threadFactory.newThread(new AsyncTask());
-        if (thread == null) {
-            throw new IllegalArgumentException("Thread factory did not create a thread");
-        }
-        thread.setDaemon(true);
-        this.queueLength = queueLength;
+        this(queueLength, threadFactory, DISCARDING_REENTRY_HANDLER);
     }
 
     /**
@@ -73,6 +71,27 @@ public AsyncHandler(final ThreadFactory threadFactory) {
         this(DEFAULT_QUEUE_LENGTH, threadFactory);
     }
 
+    /**
+     * Construct a new instance.
+     *
+     * @param reentryHandler if a sub-handler creates a log message, the log entry is passed to the consumer and not
+     *                       added to the queue
+     */
+    public AsyncHandler(final Consumer<ExtLogRecord> reentryHandler) {
+        this(DEFAULT_QUEUE_LENGTH, Executors.defaultThreadFactory(), reentryHandler);
+    }
+
+    /**
+     * Construct a new instance.
+     *
+     * @param threadFactory  the thread factory to use to construct the handler thread
+     * @param reentryHandler if a sub-handler creates a log message, the log entry is passed to the consumer and not
+     *                       added to the queue
+     */
+    public AsyncHandler(final ThreadFactory threadFactory, final Consumer<ExtLogRecord> reentryHandler) {
+        this(DEFAULT_QUEUE_LENGTH, threadFactory, reentryHandler);
+    }
+
     /**
      * Construct a new instance.
      *
@@ -82,13 +101,43 @@ public AsyncHandler(final int queueLength) {
         this(queueLength, Executors.defaultThreadFactory());
     }
 
+    /**
+     * Construct a new instance.
+     *
+     * @param queueLength    the queue length
+     * @param reentryHandler if a sub-handler creates a log message, the log entry is passed to the consumer and not
+     *                       added to the queue
+     */
+    public AsyncHandler(final int queueLength, final Consumer<ExtLogRecord> reentryHandler) {
+        this(queueLength, Executors.defaultThreadFactory(), reentryHandler);
+    }
+
     /**
      * Construct a new instance.
      */
     public AsyncHandler() {
         this(DEFAULT_QUEUE_LENGTH);
     }
 
+    /**
+     * Construct a new instance.
+     *
+     * @param queueLength    the queue length
+     * @param threadFactory  the thread factory to use to construct the handler thread
+     * @param reentryHandler if a sub-handler creates a log message, the log entry is passed to the consumer and not
+     *                       added to the queue
+     */
+    public AsyncHandler(final int queueLength, final ThreadFactory threadFactory, final Consumer<ExtLogRecord> reentryHandler) {
+        recordQueue = new ArrayBlockingQueue<>(queueLength);
+        thread = threadFactory.newThread(new AsyncTask());
+        if (thread == null) {
+            throw new IllegalArgumentException("Thread factory did not create a thread");
+        }
+        thread.setDaemon(true);
+        this.queueLength = queueLength;
+        this.reentryHandler = reentryHandler;
+    }
+
     /**
      * The full size of the queue.
      *
@@ -146,6 +195,10 @@ protected void doPublish(final ExtLogRecord record) {
             // Copy the MDC over
             record.copyMdc();
         }
+        if (JDKSpecific.isFromWorkerThread(thread, record)) {
+            reentryHandler.accept(record);
+            return;
+        }
         if (overflowAction == OverflowAction.DISCARD) {
             recordQueue.offer(record);
         } else {

src/main/java/org/jboss/logmanager/handlers/JDKSpecific.java

@@ -5,6 +5,8 @@
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 
+import org.jboss.logmanager.ExtLogRecord;
+
 /**
  * JDK-specific code relating to {@link Console}.
  */
@@ -28,4 +30,8 @@ private JDKSpecific() {
     static Charset consoleCharset() {
         return CONSOLE_CHARSET;
     }
+
+    static boolean isFromWorkerThread(final Thread worker, final ExtLogRecord record) {
+        return worker.getId() == record.getThreadID() && worker.getName().equals(record.getThreadName());
+    }
 }

src/main/java17/org/jboss/logmanager/handlers/JDKSpecific.java

@@ -5,6 +5,8 @@
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 
+import org.jboss.logmanager.ExtLogRecord;
+
 /**
  * JDK-specific code relating to {@link Console}.
  */
@@ -32,4 +34,8 @@ private JDKSpecific() {}
     static Charset consoleCharset() {
         return CONSOLE_CHARSET;
     }
+
+    static boolean isFromWorkerThread(final Thread worker, final ExtLogRecord record) {
+        return worker.getId() == record.getLongThreadID() && worker.getName().equals(record.getThreadName());
+    }
 }

src/test/java/org/jboss/logmanager/handlers/AsyncHandlerTests.java

@@ -48,7 +48,7 @@ public void setup() {
         handler = new BlockingQueueHandler();
 
         asyncHandler = new AsyncHandler();
-        asyncHandler.setOverflowAction(OverflowAction.DISCARD);
+        asyncHandler.setOverflowAction(OverflowAction.BLOCK);
         asyncHandler.addHandler(handler);
     }
 
@@ -99,6 +99,27 @@ public void testMdc() throws Exception {
         Assertions.assertEquals(mdcValue, handler.getFirst());
     }
 
+    @Test
+    public void reentry() throws Exception {
+        final ExtHandler reLogHandler = new ExtHandler() {
+
+            @Override
+            protected void doPublish(final ExtLogRecord record) {
+                super.doPublish(record);
+                // Create a new record and act is if this was through a logger
+                asyncHandler.publish(createRecord());
+            }
+        };
+        handler.addHandler(reLogHandler);
+        handler.setFormatter(new PatternFormatter("%s"));
+        asyncHandler.publish(createRecord());
+        // We should end up with one messages and a third should not happen
+        Assertions.assertEquals("Test message", handler.getFirst());
+        // This should time out. Then we end up with a null value. We could instead sleep for a shorter time and check
+        // if the queue is empty. However, a 5 second delay does not seem too long.
+        Assertions.assertNull(handler.getFirst(), () -> "Expected no more entries, but found " + handler.queue);
+    }
+
     static ExtLogRecord createRecord() {
         return new ExtLogRecord(Level.INFO, "Test message", AsyncHandlerTests.class.getName());
     }
@@ -107,13 +128,14 @@ static class BlockingQueueHandler extends ExtHandler {
         private final BlockingDeque<String> queue;
 
         BlockingQueueHandler() {
-            queue = new LinkedBlockingDeque<String>();
+            queue = new LinkedBlockingDeque<>();
             setErrorManager(AssertingErrorManager.of());
         }
 
         @Override
         protected void doPublish(final ExtLogRecord record) {
             queue.addLast(getFormatter().format(record));
+            publishToNestedHandlers(record);
             super.doPublish(record);
         }