When closing MockFluentd, wait 3 seconds until running threads for writing logs are terminated

xerial · xerial · commit fe67a1ca868d · 2014-02-25T23:17:00.000+09:00
diff --git a/src/main/java/org/fluentd/logger/sender/ExponentialDelayReconnector.java b/src/main/java/org/fluentd/logger/sender/ExponentialDelayReconnector.java
@@ -9,7 +9,6 @@
  * Calcurate exponential delay for reconnecting
  */
 public class ExponentialDelayReconnector implements Reconnector {
-    private static final Logger LOG = LoggerFactory.getLogger(ExponentialDelayReconnector.class);
 
     private double wait = 0.5;
 
diff --git a/src/test/java/org/fluentd/logger/TestFluentLogger.java b/src/test/java/org/fluentd/logger/TestFluentLogger.java
@@ -3,23 +3,47 @@
 import org.fluentd.logger.sender.Event;
 import org.fluentd.logger.sender.NullSender;
 import org.fluentd.logger.util.MockFluentd;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.msgpack.MessagePack;
 import org.msgpack.unpacker.Unpacker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.BufferedInputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.net.Socket;
 import java.util.*;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 
 public class TestFluentLogger {
+    private Logger _logger = LoggerFactory.getLogger(TestFluentLogger.class);
+
+    class FixedThreadManager {
+        private final ExecutorService service;
+
+        public FixedThreadManager(int numThreads) {
+            service = Executors.newFixedThreadPool(numThreads);
+        }
+
+        public void submit(Runnable r) {
+            service.submit(r);
+        }
+
+        public void join() throws InterruptedException {
+            service.shutdown();
+            while(!service.awaitTermination(1000, TimeUnit.MILLISECONDS)) {
+                _logger.debug("waiting ...");
+            }
+            _logger.trace("Terminating FixedThreadManager");
+        }
+    }
+
 
     @Test
     public void testNormal01() throws Exception {
@@ -42,7 +66,9 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
                 }
             }
         });
-        fluentd.start();
+
+        FixedThreadManager threadManager = new FixedThreadManager(1);
+        threadManager.submit(fluentd);
 
         // start loggers
         FluentLogger logger = FluentLogger.getLogger("testtag", host, port);
@@ -66,14 +92,15 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         fluentd.close();
 
         // wait for unpacking event data on fluentd
-        Thread.sleep(1000);
+        threadManager.join();
 
         // check data
         assertEquals(2, elist.size());
         assertEquals("testtag.test01", elist.get(0).tag);
         assertEquals("testtag.test01", elist.get(1).tag);
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testNormal02() throws Exception {
         int loggerCount = 3;
@@ -106,8 +133,8 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
                 }
             }
         });
-        ExecutorService executor = Executors.newFixedThreadPool(1);
-        executor.submit(fluentd);
+        FixedThreadManager threadManager = new FixedThreadManager(1);
+        threadManager.submit(fluentd);
 
         // start loggers
         FluentLogger[] loggers = new FluentLogger[loggerCount];
@@ -147,8 +174,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         fluentd.close();
 
         // wait for unpacking event data on fluentd
-        while(executor.awaitTermination(1000, TimeUnit.MILLISECONDS)) {
-        }
+        threadManager.join();
 
         // check data
         assertEquals(counts[0], elists[0].size());
@@ -174,6 +200,9 @@ public void testReconnection() throws Exception {
         int port = MockFluentd.randomPort();
         String host = "localhost";
         final List<Event> elist1 = new ArrayList<Event>();
+
+        FixedThreadManager threadManager = new FixedThreadManager(2);
+
         MockFluentd fluentd1 = new MockFluentd(port, new MockFluentd.MockProcess() {
             public void process(MessagePack msgpack, Socket socket) throws IOException {
                 BufferedInputStream in = new BufferedInputStream(socket.getInputStream());
@@ -192,7 +221,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
                 }
             }
         });
-        fluentd1.start();
+        threadManager.submit(fluentd1);
 
         // start loggers
         FluentLogger logger = FluentLogger.getLogger("testtag", host, port);
@@ -204,6 +233,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         }
 
         TimeUnit.MILLISECONDS.sleep(500);
+        _logger.info("Closing the current fluentd instance");
         fluentd1.closeClientSockets();
         fluentd1.close();
 
@@ -232,7 +262,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
                 }
             }
         });
-        fluentd2.start();
+        threadManager.submit(fluentd2);
 
         TimeUnit.MILLISECONDS.sleep(500);
 
@@ -245,11 +275,15 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
 
         // close loggers
         FluentLogger.closeAll();
+        Thread.sleep(2000);
 
         fluentd2.close();
 
+
         // wait for unpacking event data on fluentd
-        TimeUnit.MILLISECONDS.sleep(500);
+        TimeUnit.MILLISECONDS.sleep(2000);
+        threadManager.join();
+
 
         // check data
         assertEquals(1, elist1.size());
diff --git a/src/test/java/org/fluentd/logger/util/MockFluentd.java b/src/test/java/org/fluentd/logger/util/MockFluentd.java
@@ -6,15 +6,24 @@
 import org.msgpack.template.Templates;
 import org.msgpack.type.Value;
 import org.msgpack.unpacker.Unpacker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.HashMap;
+import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class MockFluentd extends Thread {
+
+    private static final Logger _logger = LoggerFactory.getLogger(MockFluentd.class);
+
     private ConcurrentLinkedQueue<Socket> clientSockets = new ConcurrentLinkedQueue<Socket>();
 
     public static interface MockProcess {
@@ -79,6 +88,7 @@ private static Object toObject(Unpacker u, Value v) {
 
     private MockProcess process;
 
+    private AtomicBoolean started = new AtomicBoolean(false);
     private AtomicBoolean finished = new AtomicBoolean(false);
 
     public MockFluentd(int port, MockProcess mockProcess) throws IOException {
@@ -98,35 +108,55 @@ public static int randomPort() throws IOException {
         return port;
     }
 
+    private ExecutorService service = Executors.newCachedThreadPool();
+
+
     public void run() {
+        _logger.debug("Started MockFluentd port:" + serverSocket.getLocalPort());
+
         while (!finished.get()) {
             try {
+                started.set(true);
                 final Socket socket = serverSocket.accept();
                 socket.setSoLinger(true, 0);
                 clientSockets.add(socket);
-                Runnable r = new Runnable() {
+                service.submit(new Runnable() {
                     public void run() {
                         try {
+                            _logger.trace("received log");
                             MessagePack msgpack = new MessagePack();
                             msgpack.register(Event.class, MockEventTemplate.INSTANCE);
                             process.process(msgpack, socket);
+                            _logger.trace("wrote log");
                         } catch (IOException e) {
                             // ignore
                         }
                     }
-                };
-                new Thread(r).start();
+                });
             } catch (IOException e) {
                 // ignore
             }
         }
+        _logger.debug("Terminated MockFluentd port:" + serverSocket.getLocalPort());
     }
 
     public void close() throws IOException {
         finished.set(true);
-        if (serverSocket != null) {
+        service.shutdown();
+        try {
+            // We need to wait until all log writing threads are finished.
+            int numTrial = 0;
+            while(numTrial < 3 && !service.awaitTermination(1, TimeUnit.SECONDS)) {
+                numTrial++;
+            }
+        }
+        catch(InterruptedException e) {
+            _logger.error("interrupted", e);
+        }
+         if (serverSocket != null) {
             serverSocket.close();
         }
+
     }
 
     public void closeClientSockets() {
diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml
@@ -5,11 +5,11 @@
          ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
     <!-- Meanings of the symbols are defined in http://logback.qos.ch/manual/layouts.html -->
     <encoder>
-      <pattern>%d %highlight(%-5level) [%thread] - %msg %ex{0}</pattern>
+      <pattern>%d %highlight(%-5level) [%thread] %msg%n%ex{0}</pattern>
     </encoder>
   </appender>
-
+ 
   <root level="debug">
     <appender-ref ref="STDOUT"/>
   </root>
-</configuration>
+</configuration>
