Revert "Atomic processing of create/remove of keepalive thread"

This reverts commit 4680cb8 (caused shyiko#321).

Author: shyiko

src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java

@@ -167,7 +167,6 @@ public X509Certificate[] getAcceptedIssuers() {
     private volatile ExecutorService keepAliveThreadExecutor;
 
     private final Lock connectLock = new ReentrantLock();
-    private final Lock keepAliveThreadExecutorLock = new ReentrantLock();
 
     /**
      * Alias for BinaryLogClient("localhost", 3306, <no schema> = null, username, password).
@@ -768,51 +767,46 @@ public Thread newThread(Runnable runnable) {
                     return newNamedThread(runnable, "blc-keepalive-" + hostname + ":" + port);
                 }
             });
-        try {
-            keepAliveThreadExecutorLock.lock();
-            threadExecutor.submit(new Runnable() {
-                @Override
-                public void run() {
-                    while (!threadExecutor.isShutdown()) {
+        threadExecutor.submit(new Runnable() {
+            @Override
+            public void run() {
+                while (!threadExecutor.isShutdown()) {
+                    try {
+                        Thread.sleep(keepAliveInterval);
+                    } catch (InterruptedException e) {
+                        // expected in case of disconnect
+                    }
+                    if (threadExecutor.isShutdown()) {
+                        return;
+                    }
+                    boolean connectionLost = false;
+                    if (heartbeatInterval > 0) {
+                        connectionLost = System.currentTimeMillis() - eventLastSeen > keepAliveInterval;
+                    } else {
                         try {
-                            Thread.sleep(keepAliveInterval);
-                        } catch (InterruptedException e) {
-                            // expected in case of disconnect
+                            channel.write(new PingCommand());
+                        } catch (IOException e) {
+                            connectionLost = true;
                         }
-                        if (threadExecutor.isShutdown()) {
-                            return;
-                        }
-                        boolean connectionLost = false;
-                        if (heartbeatInterval > 0) {
-                            connectionLost = System.currentTimeMillis() - eventLastSeen > keepAliveInterval;
-                        } else {
-                            try {
-                                channel.write(new PingCommand());
-                            } catch (IOException e) {
-                                connectionLost = true;
-                            }
+                    }
+                    if (connectionLost) {
+                        if (logger.isLoggable(Level.INFO)) {
+                            logger.info("Trying to restore lost connection to " + hostname + ":" + port);
                         }
-                        if (connectionLost) {
-                            if (logger.isLoggable(Level.INFO)) {
-                                logger.info("Trying to restore lost connection to " + hostname + ":" + port);
-                            }
-                            try {
-                                terminateConnect();
-                                connect(connectTimeout);
-                            } catch (Exception ce) {
-                                if (logger.isLoggable(Level.WARNING)) {
-                                    logger.warning("Failed to restore connection to " + hostname + ":" + port +
-                                        ". Next attempt in " + keepAliveInterval + "ms");
-                                }
+                        try {
+                            terminateConnect();
+                            connect(connectTimeout);
+                        } catch (Exception ce) {
+                            if (logger.isLoggable(Level.WARNING)) {
+                                logger.warning("Failed to restore connection to " + hostname + ":" + port +
+                                    ". Next attempt in " + keepAliveInterval + "ms");
                             }
                         }
                     }
                 }
-            });
-            keepAliveThreadExecutor = threadExecutor;
-        } finally {
-            keepAliveThreadExecutorLock.unlock();
-        }
+            }
+        });
+        keepAliveThreadExecutor = threadExecutor;
     }
 
     private Thread newNamedThread(Runnable runnable, String threadName) {
@@ -822,12 +816,7 @@ private Thread newNamedThread(Runnable runnable, String threadName) {
     }
 
     boolean isKeepAliveThreadRunning() {
-        try {
-            keepAliveThreadExecutorLock.lock();
-            return keepAliveThreadExecutor != null && !keepAliveThreadExecutor.isShutdown();
-        } finally {
-            keepAliveThreadExecutorLock.unlock();
-        }
+        return keepAliveThreadExecutor != null && !keepAliveThreadExecutor.isShutdown();
     }
 
     /**
@@ -1178,19 +1167,14 @@ public void disconnect() throws IOException {
     }
 
     private void terminateKeepAliveThread() {
-        try {
-            keepAliveThreadExecutorLock.lock();
-            ExecutorService keepAliveThreadExecutor = this.keepAliveThreadExecutor;
-            if (keepAliveThreadExecutor == null) {
-                return;
-            }
-            keepAliveThreadExecutor.shutdownNow();
-            while (!awaitTerminationInterruptibly(keepAliveThreadExecutor,
-                Long.MAX_VALUE, TimeUnit.NANOSECONDS)) {
-                // ignore
-            }
-        } finally {
-            keepAliveThreadExecutorLock.unlock();
+        ExecutorService keepAliveThreadExecutor = this.keepAliveThreadExecutor;
+        if (keepAliveThreadExecutor == null) {
+            return;
+        }
+        keepAliveThreadExecutor.shutdownNow();
+        while (!awaitTerminationInterruptibly(keepAliveThreadExecutor,
+            Long.MAX_VALUE, TimeUnit.NANOSECONDS)) {
+            // ignore
         }
     }