merge

CHANGELOG.md

@@ -1,4 +1,12 @@
 # Changelog
+## [0.29.2](https://github.com/osheroff/mysql-binlog-connector-java/compare/0.29.2...0.29.1) - 2024-04-22
+
+- avoid SSL deadlocks using opt-in SO_LINGER with value of 0
+
+## [0.29.1] - 2024-03-24
+
+- bug fixes for VERY long transactions
+
 ## [0.29.0](https://github.com/osheroff/mysql-binlog-connector-java/compare/0.29.0...0.28.3) - 2023-11-29
 
 - add support for mysql8 specific metadata

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>com.zendesk</groupId>
     <artifactId>mysql-binlog-connector-java</artifactId>
-    <version>0.29.0</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/event/deserialization/GtidEventDataDeserializer.java

@@ -87,7 +87,7 @@ public GtidEventData deserialize(ByteArrayInputStream inputStream) throws IOExce
                 }
                 // Total transaction length (including this GTIDEvent), introduced in MySQL-8.0.2
                 if (inputStream.available() >= TRANSACTION_LENGTH_MIN_LENGTH) {
-                    transactionLength = inputStream.readPackedInteger();
+                    transactionLength = inputStream.readPackedLong();
                 }
                 immediateServerVersion = UNDEFINED_SERVER_VERSION;
                 originalServerVersion = UNDEFINED_SERVER_VERSION;

src/main/java/com/github/shyiko/mysql/binlog/io/ByteArrayInputStream.java

@@ -159,6 +159,19 @@ public int readPackedInteger() throws IOException {
         return number.intValue();
     }
 
+    /**
+     * @see #readPackedNumber()
+     * @throws IOException in case of malformed number, eof, null
+     * @return long
+     */
+    public long readPackedLong() throws IOException {
+        Number number = readPackedNumber();
+        if (number == null) {
+            throw new IOException("Unexpected NULL where long should have been");
+        }
+        return number.longValue();
+    }
+
     /**
      * Format (first-byte-based):<br>
      * 0-250 - The first byte is the number (in the range 0-250). No additional bytes are used.<br>

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;
     }
 }