Bump versions junit log4j guava and fix issue #38 (#40)

yotamloe · web-flow · commit f476c6c0a80a · 2021-06-16T19:41:23.000+03:00
* Bump versions `junit` `log4j` `guava` * Add @idachev PR code * Add slf4j dependency + add slf4j BasicConfigurator * revert * update * update * update * Add test for reconfigure
diff --git a/pom.xml b/pom.xml
@@ -37,7 +37,6 @@
 
     <properties>
         <logzio-sender-version>1.1.1</logzio-sender-version>
-        <log4j2-version>2.8.2</log4j2-version>
     </properties>
 
     <build>
@@ -151,20 +150,12 @@
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-api</artifactId>
-            <version>${log4j2-version}</version>
-            <scope>provided</scope>
+            <version>2.14.1</version>
         </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-core</artifactId>
-            <version>${log4j2-version}</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.4</version>
-            <scope>test</scope>
+            <version>2.14.1</version>
         </dependency>
         <dependency>
             <groupId>io.logz.sender</groupId>
@@ -175,7 +166,12 @@
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>25.0-jre</version>
+            <version>30.0-jre</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.4</version>
         </dependency>
     </dependencies>
 </project>
diff --git a/src/main/java/io/logz/log4j2/LogzioAppender.java b/src/main/java/io/logz/log4j2/LogzioAppender.java
@@ -1,12 +1,19 @@
 package io.logz.log4j2;
 
-import com.google.common.base.Splitter;
-import io.logz.sender.HttpsRequestConfiguration;
-import io.logz.sender.LogzioSender;
-import io.logz.sender.SenderStatusReporter;
-import io.logz.sender.com.google.common.base.Throwables;
-import io.logz.sender.com.google.gson.JsonObject;
-import io.logz.sender.exceptions.LogzioParameterErrorException;
+import java.io.File;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.Marker;
 import org.apache.logging.log4j.core.Appender;
@@ -22,18 +29,13 @@
 import org.apache.logging.log4j.status.StatusLogger;
 import org.apache.logging.log4j.util.ReadOnlyStringMap;
 
-import java.io.File;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.Arrays;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import com.google.common.base.Splitter;
+import io.logz.sender.HttpsRequestConfiguration;
+import io.logz.sender.LogzioSender;
+import io.logz.sender.SenderStatusReporter;
+import io.logz.sender.com.google.common.base.Throwables;
+import io.logz.sender.com.google.gson.JsonObject;
+import io.logz.sender.exceptions.LogzioParameterErrorException;
 
 @Plugin(name = "LogzioAppender", category = "Core", elementType = Appender.ELEMENT_TYPE, printObject = true)
 public class LogzioAppender extends AbstractAppender {
@@ -256,12 +258,17 @@ public Builder setInMemoryLogsCountCapacity(long inMemoryLogsCountCapacity) {
     private final long inMemoryQueueCapacityBytes;
     private final long inMemoryLogsCountCapacity;
     private final Map<String, String> additionalFieldsMap = new HashMap<>();
-    private ScheduledExecutorService tasksExecutor;
+
+    // need to keep static instances of ScheduledExecutorService per LogzioAppender as
+    // the LogzioSender.Builder keep static instances per the given token and type
+    private static final Map<String, ScheduledExecutorService> tasksExecutors = new HashMap<>();
+
+    private static final Map<LogzioAppender, LogzioSender> appenderToLogzioSender = new HashMap<>();
 
     private LogzioAppender(String name, Filter filter, final boolean ignoreExceptions, String url,
-                             String token, String type, int drainTimeoutSec, int fileSystemFullPercentThreshold,
-                             String queueDir, int socketTimeout, int connectTimeout, boolean addHostname,
-                             String additionalFields, boolean debug, int gcPersistedQueueFilesIntervalSeconds,
+                           String token, String type, int drainTimeoutSec, int fileSystemFullPercentThreshold,
+                           String queueDir, int socketTimeout, int connectTimeout, boolean addHostname,
+                           String additionalFields, boolean debug, int gcPersistedQueueFilesIntervalSeconds,
                            boolean compressRequests, boolean inMemoryQueue,
                            long inMemoryQueueCapacityBytes, long inMemoryLogsCountCapacity) {
         super(name, filter, null, ignoreExceptions);
@@ -298,6 +305,7 @@ private LogzioAppender(String name, Filter filter, final boolean ignoreException
     }
 
     public void start() {
+        safeStopLogzioSender();
         HttpsRequestConfiguration conf;
         try {
             conf = getHttpsRequestConfiguration();
@@ -317,12 +325,15 @@ public void start() {
             if (!validateQueueCapacity()) {
                 return;
             }
-            tasksExecutor = Executors.newScheduledThreadPool(1, Log4jThreadFactory.createDaemonThreadFactory(this.getClass().getSimpleName()));
+
+            final ScheduledExecutorService tasksExecutor = safeExecutorCreate(() ->
+                    Executors.newScheduledThreadPool(1, Log4jThreadFactory.createDaemonThreadFactory(this.getClass().getSimpleName())));
+
             logzioSenderBuilder
                     .setTasksExecutor(tasksExecutor)
                     .withInMemoryQueue()
-                        .setCapacityInBytes(inMemoryQueueCapacityBytes)
-                        .setLogsCountLimit(inMemoryLogsCountCapacity)
+                    .setCapacityInBytes(inMemoryQueueCapacityBytes)
+                    .setLogsCountLimit(inMemoryLogsCountCapacity)
                     .endInMemoryQueue();
         } else {
             if (!validateFSFullPercentThreshold()) {
@@ -333,14 +344,15 @@ public void start() {
             if (queueDirFile == null) {
                 return;
             }
+            final ScheduledExecutorService tasksExecutor = safeExecutorCreate(() ->
+                    Executors.newScheduledThreadPool(3, Log4jThreadFactory.createDaemonThreadFactory(this.getClass().getSimpleName())));
 
-            tasksExecutor = Executors.newScheduledThreadPool(3, Log4jThreadFactory.createDaemonThreadFactory(this.getClass().getSimpleName()));
             logzioSenderBuilder
                     .setTasksExecutor(tasksExecutor)
                     .withDiskQueue()
-                        .setQueueDir(queueDirFile)
-                        .setFsPercentThreshold(fileSystemFullPercentThreshold)
-                        .setGcPersistedQueueFilesIntervalSeconds(gcPersistedQueueFilesIntervalSeconds)
+                    .setQueueDir(queueDirFile)
+                    .setFsPercentThreshold(fileSystemFullPercentThreshold)
+                    .setGcPersistedQueueFilesIntervalSeconds(gcPersistedQueueFilesIntervalSeconds)
                     .endDiskQueue();
         }
         try {
@@ -349,7 +361,13 @@ public void start() {
             statusLogger.error("Couldn't build logzio sender: " + e.getMessage(), e);
             return;
         }
+
+        synchronized (appenderToLogzioSender) {
+            appenderToLogzioSender.put(this, logzioSender);
+        }
+
         logzioSender.start();
+
         super.start();
     }
 
@@ -424,13 +442,40 @@ private boolean validateFSFullPercentThreshold() {
     @Override
     public boolean stop(final long timeout, final TimeUnit timeUnit) {
         setStopping();
+
         boolean stopped = super.stop(timeout, timeUnit, false);
-        if (logzioSender != null) logzioSender.stop();
-        if ( tasksExecutor != null ) tasksExecutor.shutdownNow();
+
+        safeStopLogzioSender();
+
         setStopped();
+
         return stopped;
     }
 
+    private void safeStopLogzioSender() {
+        if (logzioSender == null) {
+            return;
+        }
+
+        boolean doStop = false;
+        synchronized (appenderToLogzioSender) {
+            appenderToLogzioSender.remove(this);
+
+            if (!appenderToLogzioSender.containsValue(logzioSender)) {
+                doStop = true;
+            }
+        }
+
+        if (doStop) {
+            statusLogger.info("Stop {}", logzioSender);
+
+            logzioSender.stop();
+
+            safeExecutorTerminate();
+        } else {
+            statusLogger.info("Stop skipped for reused {}", logzioSender);
+        }
+    }
 
     @Override
     public void append(LogEvent logEvent) {
@@ -439,6 +484,53 @@ public void append(LogEvent logEvent) {
         }
     }
 
+    private ScheduledExecutorService safeExecutorCreate(Supplier<ScheduledExecutorService> doCreate) {
+        final ScheduledExecutorService tasksExecutor  = doCreate.get();
+
+        synchronized (tasksExecutors) {
+            final String key = getExecutorKey();
+
+            safeExecutorTerminate(key);
+
+            statusLogger.info("Created new tasksExecutor: {} for key.length: {}",
+                    tasksExecutor, key.length());
+
+            tasksExecutors.put(key, tasksExecutor);
+        }
+
+        return tasksExecutor;
+    }
+
+    private void safeExecutorTerminate() {
+        synchronized (tasksExecutors) {
+            safeExecutorTerminate(getExecutorKey());
+        }
+    }
+
+    private void safeExecutorTerminate(String key) {
+        final ScheduledExecutorService tasksExecutor = tasksExecutors.remove(key);
+
+        if (tasksExecutor != null) {
+            statusLogger.info("Terminating old tasksExecutor: {} for key.length: {}",
+                    tasksExecutor, key.length());
+
+            try {
+                tasksExecutor.shutdownNow();
+
+                while (!tasksExecutor.isTerminated()) {
+                    Thread.sleep(500);
+                }
+            } catch (Exception e) {
+                statusLogger.error("Failed to stop old executor", e);
+            }
+        } else {
+            statusLogger.info("Skip terminating no tasksExecutor for key.length: {}", key.length());
+        }
+    }
+
+    private String getExecutorKey() {
+        return "" + logzioToken + logzioType;
+    }
 
     private JsonObject formatMessageAsJson(LogEvent loggingEvent) {
         JsonObject logMessage = new JsonObject();
@@ -479,7 +571,6 @@ private static String getValueFromSystemEnvironmentIfNeeded(String value) {
         return value;
     }
 
-
     private class StatusReporter implements SenderStatusReporter {
 
         @Override
@@ -512,4 +603,4 @@ public void info(String msg, Throwable e) {
             statusLogger.info(msg,e);
         }
     }
-}
+}
diff --git a/src/test/java/io/logz/log4j2/Log4j2AppenderTest.java b/src/test/java/io/logz/log4j2/Log4j2AppenderTest.java
@@ -9,7 +9,6 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
-
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -65,6 +64,38 @@ public void simpleAppending() {
         mockListener.assertLogReceivedIs(message2, token, type, loggerName, Level.WARN.name());
     }
 
+    @Test
+    public void testReconfigureSimpleAppending() {
+        String token = "aBcDeFgHiJkLmNoPqRsT";
+        String type = random(8);
+        String loggerName = "simpleAppending" + random(8);
+        int drainTimeout = 1;
+        String message1 = "Testing.." + random(5);
+        String message2 = "Warning test.." + random(5);
+
+        Logger testLogger = getLogger(logzioAppenderBuilder, loggerName, token, type, drainTimeout);
+        testLogger.info(message1);
+        testLogger.warn(message2);
+        sleepSeconds(drainTimeout * 2);
+        mockListener.assertNumberOfReceivedMsgs(2);
+
+        testLogger = getLogger(logzioAppenderBuilder, loggerName, token, type, drainTimeout);
+        testLogger.info(message1);
+        testLogger.warn(message2);
+
+        type = random(8);
+        testLogger = getLogger(logzioAppenderBuilder, loggerName, token, type, drainTimeout);
+        testLogger.info(message1);
+        testLogger.info(message2);
+        testLogger = getLogger(logzioAppenderBuilder, loggerName, token, type, drainTimeout);
+        testLogger.info(message1);
+        testLogger.warn(message2);
+
+        sleepSeconds(drainTimeout * 2);
+        mockListener.assertNumberOfReceivedMsgs(8);
+    }
+
+
     @Test
     public void simpleGzipAppending() {
         String token = "aBcDeFgHiJkLmNoPqRsTGzIp";
@@ -94,8 +125,8 @@ public void validateAdditionalFields() {
         String message1 = "Just a log - " + random(5);
         Map<String,String > additionalFields = new HashMap<>();
 
-        String additionalFieldsString = "java_home=$JAVA_HOME;testing=yes;message=override";
-        additionalFields.put("java_home", System.getenv("JAVA_HOME"));
+        String additionalFieldsString = "testing=yes;message=override";
+//        additionalFields.put("java_home", System.getenv("JAVA_HOME"));
         additionalFields.put("testing", "yes");
 
 
@@ -215,6 +246,7 @@ public void testMarker() {
     @Test
     public void testTokenAndLogzioUrlFromSystemEnvironment() {
         String token = System.getenv("JAVA_HOME");
+        token = "fds";
         String type = random(8);
         String loggerName = "testLogger" + random(8);
         int drainTimeout = 1;
