Add timeout to wait until queued messages are processed

Author: snorwin

src/main/java/com/splunk/logging/HttpEventCollectorLog4jAppender.java

@@ -157,7 +157,7 @@ public static HttpEventCollectorLog4jAppender createAppender(
             @PluginAttribute(value = "call_timeout", defaultLong = HttpEventCollectorSender.TimeoutSettings.DEFAULT_CALL_TIMEOUT) final long callTimeout,
             @PluginAttribute(value = "read_timeout", defaultLong = HttpEventCollectorSender.TimeoutSettings.DEFAULT_READ_TIMEOUT) final long readTimeout,
             @PluginAttribute(value = "write_timeout", defaultLong = HttpEventCollectorSender.TimeoutSettings.DEFAULT_WRITE_TIMEOUT) final long writeTimeout,
-            @PluginAttribute(value = "await_termination_timeout", defaultLong = HttpEventCollectorSender.TimeoutSettings.DEFAULT_AWAIT_TERMINATION_TIMEOUT) final long awaitTerminationTimeout,
+            @PluginAttribute(value = "termination_timeout", defaultLong = HttpEventCollectorSender.TimeoutSettings.DEFAULT_TERMINATION_TIMEOUT) final long terminationTimeout,
             @PluginElement("Layout") Layout<? extends Serializable> layout,
             @PluginElement("Filter") final Filter filter
     )
@@ -220,7 +220,7 @@ public static HttpEventCollectorLog4jAppender createAppender(
                 disableCertificateValidation,
                 eventBodySerializer,
                 eventHeaderSerializer,
-                new HttpEventCollectorSender.TimeoutSettings(connectTimeout, callTimeout, readTimeout, writeTimeout, awaitTerminationTimeout)
+                new HttpEventCollectorSender.TimeoutSettings(connectTimeout, callTimeout, readTimeout, writeTimeout, terminationTimeout)
         );
     }
 

src/main/java/com/splunk/logging/HttpEventCollectorLogbackAppender.java

@@ -383,12 +383,12 @@ public long getWriteTimeout(long milliseconds) {
         return this.timeoutSettings.writeTimeout = milliseconds;
     }
 
-    public void setAwaitTerminationTimeout(long milliseconds) {
-        this.timeoutSettings.awaitTerminationTimeout = milliseconds;
+    public void setTerminationTimeout(long milliseconds) {
+        this.timeoutSettings.terminationTimeout = milliseconds;
     }
 
-    public long getAwaitTerminationTimeout(long milliseconds) {
-        return this.timeoutSettings.awaitTerminationTimeout = milliseconds;
+    public long getTerminationTimeout(long milliseconds) {
+        return this.timeoutSettings.terminationTimeout = milliseconds;
     }
 
     private static long parseLong(String string, int defaultValue) {

src/main/java/com/splunk/logging/HttpEventCollectorLoggingHandler.java

@@ -114,7 +114,7 @@ public final class HttpEventCollectorLoggingHandler extends Handler {
     private final String CallTimeoutConfTag = "call_timeout";
     private final String ReadTimeoutConfTag = "read_timeout";
     private final String WriteTimeoutConfTag = "write_timeout";
-    private final String AwaitTerminationTimeoutConfTag = "await_termination_timeout";
+    private final String TerminationTimeoutConfTag = "termination_timeout";
 
     /** HttpEventCollectorLoggingHandler c-or */
     public HttpEventCollectorLoggingHandler() {
@@ -167,7 +167,7 @@ public HttpEventCollectorLoggingHandler() {
             getConfigurationNumericProperty(CallTimeoutConfTag, HttpEventCollectorSender.TimeoutSettings.DEFAULT_CALL_TIMEOUT),
             getConfigurationNumericProperty(ReadTimeoutConfTag, HttpEventCollectorSender.TimeoutSettings.DEFAULT_READ_TIMEOUT),
             getConfigurationNumericProperty(WriteTimeoutConfTag, HttpEventCollectorSender.TimeoutSettings.DEFAULT_WRITE_TIMEOUT),
-            getConfigurationNumericProperty(AwaitTerminationTimeoutConfTag, HttpEventCollectorSender.TimeoutSettings.DEFAULT_AWAIT_TERMINATION_TIMEOUT)
+            getConfigurationNumericProperty(TerminationTimeoutConfTag, HttpEventCollectorSender.TimeoutSettings.DEFAULT_TERMINATION_TIMEOUT)
         );
 
         if ("raw".equalsIgnoreCase(type)) {

src/main/java/com/splunk/logging/HttpEventCollectorSender.java

@@ -263,15 +263,36 @@ public static void putIfPresent(JsonObject collection, String tag, Object value)
 
     private void stopHttpClient() {
         if (httpClient != null) {
-            httpClient.dispatcher().executorService().shutdown();
+            Dispatcher dispatcher = httpClient.dispatcher();
+            httpClient = null;
 
-            if (timeoutSettings.awaitTerminationTimeout > 0) {
-                try {
-                    httpClient.dispatcher().executorService().awaitTermination(timeoutSettings.awaitTerminationTimeout, TimeUnit.MILLISECONDS);
-                } catch (InterruptedException ignored) { /* nop */ }
-            }
+            if (timeoutSettings.terminationTimeout > 0) {
+                // wait for queued messages in the dispatcher to be promoted to the executor service
+                long start = System.currentTimeMillis();
+                while (dispatcher.queuedCallsCount() > 0 && start + timeoutSettings.terminationTimeout > System.currentTimeMillis()) {
+                    try {
+                        TimeUnit.MILLISECONDS.sleep(10);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        break;
+                    }
+                }
 
-            httpClient = null;
+                // initialize the shutdown of the executor service
+                dispatcher.executorService().shutdown();
+
+                // wait for the messages in the dispatcher's executor service to be sent out
+                long awaitTerminationTimeout = timeoutSettings.terminationTimeout - (System.currentTimeMillis() - start);
+                if (awaitTerminationTimeout > 0) {
+                    try {
+                        dispatcher.executorService().awaitTermination(awaitTerminationTimeout, TimeUnit.MILLISECONDS);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            } else {
+                dispatcher.executorService().shutdown();
+            }
         }
     }
 
@@ -400,22 +421,22 @@ public static class TimeoutSettings {
         public static final long DEFAULT_WRITE_TIMEOUT = 0; // 0 means no timeout
         public static final long DEFAULT_CALL_TIMEOUT = 0;
         public static final long DEFAULT_READ_TIMEOUT = 0;
-        public static final long DEFAULT_AWAIT_TERMINATION_TIMEOUT = 0;
+        public static final long DEFAULT_TERMINATION_TIMEOUT = 0;
 
         public long connectTimeout = DEFAULT_CONNECT_TIMEOUT;
         public long callTimeout = DEFAULT_CALL_TIMEOUT;
         public long readTimeout = DEFAULT_READ_TIMEOUT;
         public long writeTimeout = DEFAULT_WRITE_TIMEOUT;
-        public long awaitTerminationTimeout = DEFAULT_AWAIT_TERMINATION_TIMEOUT;
+        public long terminationTimeout = DEFAULT_TERMINATION_TIMEOUT;
 
         public TimeoutSettings() {}
 
-        public TimeoutSettings(long connectTimeout, long callTimeout, long readTimeout, long writeTimeout, long awaitTerminationTimeout) {
+        public TimeoutSettings(long connectTimeout, long callTimeout, long readTimeout, long writeTimeout, long terminationTimeout) {
             this.connectTimeout = connectTimeout;
             this.callTimeout = callTimeout;
             this.readTimeout = readTimeout;
             this.writeTimeout = writeTimeout;
-            this.awaitTerminationTimeout = awaitTerminationTimeout;
+            this.terminationTimeout = terminationTimeout;
         }
     }
 }

src/test/resources/log4j2.xml

@@ -45,7 +45,7 @@ under the License.
               batch_size_count="0"
               batch_interval="0"
               connect_timeout="5000"
-              await_termination_timeout="1000"
+              termination_timeout="1000"
               disableCertificateValidation="true">
             <PatternLayout pattern="%m"/>
         </SplunkHttp>

src/test/resources/logback.xml

@@ -60,7 +60,7 @@ under the License.
         <messageFormat>text</messageFormat>
         <middleware>HttpEventCollectorUnitTestMiddleware</middleware>
         <connectTimeout>5000</connectTimeout>
-        <awaitTerminationTieout>2000</awaitTerminationTieout>
+        <terminationTieout>2000</terminationTieout>
         <layout class="ch.qos.logback.classic.PatternLayout">
             <pattern>%msg</pattern>
         </layout>