Avoid Thread Leak when used dynamically in a Route (#143)

* WIP testing * Cleanup * Set success on failure
bluedenim · Feb 28, 2024 · d971d81 · d971d81
1 parent 1331913
commit d971d81
Show file tree

Hide file tree

Showing 4 changed files with 104 additions and 34 deletions.
diff --git a/appender-core/src/main/java/com/van/logging/LoggingEventCache.java b/appender-core/src/main/java/com/van/logging/LoggingEventCache.java
@@ -1,10 +1,19 @@
 package com.van.logging;
 
-import org.apache.logging.log4j.core.LogEvent;
-
-import java.io.*;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.util.Queue;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -67,25 +76,7 @@ public static boolean shutDown() throws InterruptedException {
         LoggingEventCache instance = instances.poll();
         while (null != instance) {
             try {
-                ExecutorService executorService =
-                    (ExecutorService) instance.executorServiceRef.getAndSet(null);
-                if (null != executorService) {
-                    if (instance.verbose) {
-                        VansLogger.logger.info(String.format("LoggingEventCache %s: shutting down", instance));
-                    }
-                    executorService.shutdown();
-                    boolean terminated = executorService.awaitTermination(
-                        SHUTDOWN_TIMEOUT_SECS,
-                        TimeUnit.SECONDS
-                    );
-                    if (instance.verbose) {
-                        VansLogger.logger.info(String.format(
-                            "LoggingEventCache: Executor service terminated within timeout: %s", terminated
-                        ));
-                    }
-                    success = success & terminated;
-                }
-
+                instance.stop();
                 if (null != instance.cacheMonitor) {
                     instance.cacheMonitor.shutDown();
                 }
@@ -266,5 +257,34 @@ Future<Boolean> publishCache(final String name, final boolean useCurrentThread)
         }
         return CompletableFuture.completedFuture(success);
     }
+
+    public boolean stop() {
+        boolean success = true;
+        final ExecutorService executorService = executorServiceRef.getAndSet(null);
+
+        if (executorService != null && !executorService.isShutdown()) {
+            try {
+                if (verbose) {
+                    VansLogger.logger.info(String.format("LoggingEventCache %s: shutting down", executorService));
+                }
+                executorService.shutdown();
+                boolean terminated = executorService.awaitTermination(
+                        SHUTDOWN_TIMEOUT_SECS,
+                        TimeUnit.SECONDS
+                );
+                if (verbose) {
+                    VansLogger.logger.info(String.format(
+                            "LoggingEventCache: Executor service terminated within timeout: %s", terminated
+                    ));
+                }
+                success = success & terminated;
+            } catch (Exception ex) {
+                VansLogger.logger.error(String.format("LoggingEventCache: error shutting down %s", executorService), ex);
+                success = false;
+            }
+        }
+
+        return success;
+    }
 }
 
diff --git a/appender-log4j2/src/main/java/com/van/logging/log4j2/Log4j2Appender.java b/appender-log4j2/src/main/java/com/van/logging/log4j2/Log4j2Appender.java
@@ -7,7 +7,8 @@
 import org.apache.logging.log4j.core.Layout;
 import org.apache.logging.log4j.core.LogEvent;
 import org.apache.logging.log4j.core.appender.AbstractAppender;
-import org.apache.logging.log4j.core.config.plugins.*;
+import org.apache.logging.log4j.core.config.plugins.Plugin;
+import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
 
 import java.io.Serializable;
 import java.util.Objects;
@@ -37,16 +38,9 @@ public static org.apache.logging.log4j.core.util.Builder<Log4j2Appender> newBuil
         super(name, filter, layout, ignoreExceptions);
         Objects.requireNonNull(eventCache);
         this.eventCache = eventCache;
-
         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-            if (this.verbose) {
-                VansLogger.logger.info("Publishing staging log on shutdown...");
-            }
-            eventCache.flushAndPublish(true);
             try {
-                if (this.verbose) {
-                    VansLogger.logger.info("Shutting down LoggingEventCache...");
-                }
+                close();
                 LoggingEventCache.shutDown();
             } catch (InterruptedException e) {
                 if (this.verbose) {
@@ -61,6 +55,7 @@ public Log4j2Appender setVerbose(boolean verbose) {
         return this;
     }
 
+    @Override
     public void append(LogEvent logEvent) {
         try {
             eventCache.add(mapToEvent(logEvent));
@@ -74,6 +69,12 @@ public void append(LogEvent logEvent) {
         }
     }
 
+    @Override
+    public void stop() {
+        close();
+        super.stop();
+    }
+
     Event mapToEvent(LogEvent event) {
         String message = null;
         if (null != getLayout()) {
@@ -88,4 +89,15 @@ Event mapToEvent(LogEvent event) {
         );
         return mapped;
     }
+
+    /**
+     * Closes resources used by this appender instance.
+     */
+    private void close() {
+        if (this.verbose) {
+            VansLogger.logger.info("Publishing staging log on close...");
+        }
+        eventCache.flushAndPublish(true);
+        eventCache.stop();
+    }
 }
diff --git a/appender-log4j2/src/main/java/com/van/logging/log4j2/Log4j2AppenderBuilder.java b/appender-log4j2/src/main/java/com/van/logging/log4j2/Log4j2AppenderBuilder.java
@@ -1,6 +1,12 @@
 package com.van.logging.log4j2;
 
-import com.van.logging.*;
+import com.van.logging.BufferPublisher;
+import com.van.logging.CapacityBasedBufferMonitor;
+import com.van.logging.IBufferMonitor;
+import com.van.logging.IBufferPublisher;
+import com.van.logging.LoggingEventCache;
+import com.van.logging.TimePeriodBasedBufferMonitor;
+import com.van.logging.VansLogger;
 import com.van.logging.aws.S3Configuration;
 import com.van.logging.aws.S3PublishHelper;
 import com.van.logging.azure.BlobConfiguration;
@@ -18,7 +24,11 @@
 import org.apache.logging.log4j.core.filter.AbstractFilter;
 
 import java.net.UnknownHostException;
-import java.util.*;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 

diff --git a/appender-log4j2/src/test/java/com/van/logging/log4j2/Log4j2AppenderTest.java b/appender-log4j2/src/test/java/com/van/logging/log4j2/Log4j2AppenderTest.java
@@ -0,0 +1,28 @@
+package com.van.logging.log4j2;
+
+import com.van.logging.LoggingEventCache;
+import junit.framework.TestCase;
+import org.apache.logging.log4j.core.Filter;
+import org.apache.logging.log4j.core.Layout;
+
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.mock;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+
+public class Log4j2AppenderTest extends TestCase {
+
+    public void testClose() {
+        final String name = "test";
+        final Filter filter = mock(Filter.class);
+        final Layout layout = mock(Layout.class);
+        final LoggingEventCache loggingEventCache = mock(LoggingEventCache.class);
+        final Log4j2Appender appender = new Log4j2Appender(name, filter, layout, false, loggingEventCache);
+
+        expect(loggingEventCache.flushAndPublish(true)).andReturn(null);
+        expect(loggingEventCache.stop()).andReturn(true);
+        replay(loggingEventCache);
+        appender.stop();
+        verify(loggingEventCache);
+    }
+}