DBZ-7570/osheroff#133: add workaround using SO_LINGER with 0 timeout

Author: acristu

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>com.zendesk</groupId>
     <artifactId>mysql-binlog-connector-java</artifactId>
-    <version>0.29.1</version>
+    <version>0.29.2</version>
 
     <name>mysql-binlog-connector-java</name>
     <description>MySQL Binary Log connector</description>

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

@@ -137,6 +137,7 @@ public X509Certificate[] getAcceptedIssuers() {
     private volatile long binlogPosition = 4;
     private volatile long connectionId;
     private SSLMode sslMode = SSLMode.DISABLED;
+    private boolean useNonGracefulDisconnect = false;
 
     protected GtidSet gtidSet;
     protected final Object gtidSetAccessLock = new Object();
@@ -249,6 +250,10 @@ public void setSSLMode(SSLMode sslMode) {
         this.sslMode = sslMode;
     }
 
+    public void setUseNonGracefulDisconnect(boolean useNonGracefulDisconnect) {
+        this.useNonGracefulDisconnect = useNonGracefulDisconnect;
+    }
+
     public long getMasterServerId() {
         return this.masterServerId;
     }
@@ -891,7 +896,7 @@ public void run() {
                         if (connectionLost) {
                             logger.info("Keepalive: Trying to restore lost connection to " + hostname + ":" + port);
                             try {
-                                terminateConnect();
+                                terminateConnect(useNonGracefulDisconnect);
                                 connect(connectTimeout);
                             } catch (Exception ce) {
                                 logger.warning("keepalive: Failed to restore connection to " + hostname + ":" + port +
@@ -1341,8 +1346,11 @@ private static boolean awaitTerminationInterruptibly(ExecutorService executorSer
     }
 
     private void terminateConnect() throws IOException {
+        terminateConnect(false);
+    }
+    private void terminateConnect(boolean force) throws IOException {
         do {
-            disconnectChannel();
+            disconnectChannel(force);
         } while (!tryLockInterruptibly(connectLock, 1000, TimeUnit.MILLISECONDS));
         connectLock.unlock();
     }
@@ -1356,8 +1364,14 @@ private static boolean tryLockInterruptibly(Lock lock, long time, TimeUnit unit)
     }
 
     private void disconnectChannel() throws IOException {
+        disconnectChannel(false);
+    }
+    private void disconnectChannel(boolean force) throws IOException {
         connected = false;
         if (channel != null && channel.isOpen()) {
+            if (force) {
+                channel.setShouldUseSoLinger0();
+            }
             channel.close();
         }
     }

src/main/java/com/github/shyiko/mysql/binlog/network/protocol/PacketChannel.java

@@ -26,6 +26,7 @@
 import javax.net.ssl.SSLSocket;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketException;
 import java.nio.channels.Channel;
 
 /**
@@ -38,6 +39,7 @@ public class PacketChannel implements Channel {
     private Socket socket;
     private ByteArrayInputStream inputStream;
     private ByteArrayOutputStream outputStream;
+    private boolean shouldUseSoLinger0 = false;
 
     public PacketChannel(String hostname, int port) throws IOException {
         this(new Socket(hostname, port));
@@ -109,6 +111,10 @@ public boolean isSSL() {
         return isSSL;
     }
 
+    public void setShouldUseSoLinger0() {
+        shouldUseSoLinger0 = true;
+    }
+
     @Override
     public boolean isOpen() {
         return !socket.isClosed();
@@ -126,6 +132,14 @@ public void close() throws IOException {
         } catch (Exception e) {
             // ignore
         }
+        if (shouldUseSoLinger0) {
+            try {
+                socket.setSoLinger(true, 0);
+            } catch (SocketException e) {
+                // ignore
+            }
+        }
         socket.close();
+        shouldUseSoLinger0 = false;
     }
 }