Merge pull request #4 from alibaba/master

merge from master

Author: zhangjiayin

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java

@@ -30,18 +30,17 @@
  */
 public class MysqlConnector {
 
+    public static final int     timeout           = 5 * 1000;                                     // 5s
     private static final Logger logger            = LoggerFactory.getLogger(MysqlConnector.class);
     private InetSocketAddress   address;
     private String              username;
     private String              password;
     private SslInfo             sslInfo;
-
     private String              defaultSchema;
     private int                 soTimeout         = 30 * 1000;
     private int                 connTimeout       = 5 * 1000;
     private int                 receiveBufferSize = 16 * 1024;
     private int                 sendBufferSize    = 16 * 1024;
-
     private SocketChannel       channel;
     private volatile boolean    dumping           = false;
     // mysql connectionId
@@ -50,8 +49,6 @@ public class MysqlConnector {
     // serverVersion
     private String              serverVersion;
 
-    public static final int     timeout           = 5 * 1000;                                     // 5s
-
     public MysqlConnector(){
     }
 
@@ -200,14 +197,15 @@ private void negotiate(SocketChannel channel) throws IOException {
         }
         HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
         handshakePacket.fromBytes(body);
+        byte serverCharsetNumber = (handshakePacket.serverCharsetNumber != 0) ? handshakePacket.serverCharsetNumber : 33;
         SslMode sslMode = sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED;
         if (sslMode != SslMode.DISABLED) {
             boolean serverSupportSsl = (handshakePacket.serverCapabilities & CLIENT_SSL) > 0;
             if (!serverSupportSsl) {
                 throw new IOException("MySQL Server does not support SSL: " + address + " serverCapabilities: "
                                       + handshakePacket.serverCapabilities);
             }
-            byte[] sslPacket = new SslRequestCommandPacket(handshakePacket.serverCharsetNumber).toBytes();
+            byte[] sslPacket = new SslRequestCommandPacket(serverCharsetNumber).toBytes();
             HeaderPacket sslHeader = new HeaderPacket();
             sslHeader.setPacketBodyLength(sslPacket.length);
             sslHeader.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
@@ -225,23 +223,27 @@ private void negotiate(SocketChannel channel) throws IOException {
         connectionId = handshakePacket.threadId; // 记录一下connection
         serverVersion = handshakePacket.serverVersion; // 记录serverVersion
         logger.info("handshake initialization packet received, prepare the client authentication packet to send");
-        logger.info("auth plugin: {}", new String(handshakePacket.authPluginName));
+        // 某些老协议的 server 默认不返回 auth plugin，需要使用默认的 mysql_native_password
+        String authPluginName = (handshakePacket.authPluginName != null
+                                 && handshakePacket.authPluginName.length > 0) ? new String(
+                                     handshakePacket.authPluginName) : "mysql_native_password";
+        logger.info("auth plugin: {}", authPluginName);
         boolean isSha2Password = false;
         ClientAuthenticationPacket clientAuth;
-        if ("caching_sha2_password".equals(new String(handshakePacket.authPluginName))) {
+        if ("caching_sha2_password".equals(authPluginName)) {
             clientAuth = new ClientAuthenticationSHA2Packet();
             isSha2Password = true;
         } else {
             clientAuth = new ClientAuthenticationPacket();
         }
-        clientAuth.setCharsetNumber(handshakePacket.serverCharsetNumber);
+        clientAuth.setCharsetNumber(serverCharsetNumber);
 
         clientAuth.setUsername(username);
         clientAuth.setPassword(password);
         clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);
         clientAuth.setDatabaseName(defaultSchema);
         clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket));
-        clientAuth.setAuthPluginName(handshakePacket.authPluginName);
+        clientAuth.setAuthPluginName(authPluginName.getBytes());
 
         byte[] clientAuthPkgBody = clientAuth.toBytes();
         HeaderPacket h = new HeaderPacket();
@@ -284,7 +286,7 @@ private void negotiate(SocketChannel channel) throws IOException {
                     encryptedPassword = getPassword().getBytes();
                     header = authSwitchAfterAuth(encryptedPassword, header);
                     body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
-                } else if ("mysql_native_password".equals(pluginName)) {
+                } else if (pluginName == null || "mysql_native_password".equals(pluginName)) {
                     try {
                         encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData);
                     } catch (NoSuchAlgorithmException e) {
@@ -481,10 +483,6 @@ public void setChannel(SocketChannel channel) {
         this.channel = channel;
     }
 
-    public void setPassword(String password) {
-        this.password = password;
-    }
-
     public long getConnectionId() {
         return connectionId;
     }
@@ -513,6 +511,10 @@ public String getPassword() {
         return password;
     }
 
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
     public String getServerVersion() {
         return serverVersion;
     }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java

@@ -167,16 +167,6 @@ public SocketAddress getLocalSocketAddress() {
     public void close() {
         Socket socket = this.socket;
         if (socket != null) {
-            try {
-                socket.shutdownInput();
-            } catch (IOException e) {
-                // Ignore, could not do anymore
-            }
-            try {
-                socket.shutdownOutput();
-            } catch (IOException e) {
-                // Ignore, could not do anymore
-            }
             try {
                 socket.close();
             } catch (IOException e) {

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java

@@ -109,7 +109,7 @@ private static SSLSocket createSslSocket(Socket socket, SslInfo sslInfo) throws
         }
         logger.info("SSL protocol: {}", StringUtils.join(protocolArr, ","));
         sslSocket.setEnabledProtocols(protocolArr);
-        // sslSocket.setEnabledCipherSuites("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_DSS_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_DSS_WITH_AES_128_GCM_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384,TLS_RSA_WITH_AES_256_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA384,TLS_DHE_RSA_WITH_AES_256_CBC_SHA256,TLS_DHE_DSS_WITH_AES_256_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA,TLS_DHE_RSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256,TLS_RSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_DSS_WITH_AES_128_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_128_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_DSS_WITH_AES_128_CBC_SHA".split(","));
+        sslSocket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
         sslSocket.startHandshake();
         logger.info("SSL socket handshake success.");
         return sslSocket;

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java

@@ -603,6 +603,32 @@ private void loadVersionComment() {
         }
     }
 
+    /**
+     * MySQL 8.4版本开始部分命令出现变化
+     * https://dev.mysql.com/doc/relnotes/mysql/8.4/en/news-8-4-0.html#mysqld-8-4-0-deprecation-removal
+     * 
+     * @param major
+     * @param minor
+     * @return
+     */
+    public boolean atLeast(int major, int minor) {
+        if (isMariaDB()) {
+            return false;
+        }
+        String version = connector.getServerVersion();
+        if (StringUtils.isNotEmpty(version)) {
+            String[] parts = version.split("\\.");
+            int majorVer = Integer.parseInt(parts[0]);
+            int minorVer = Integer.parseInt(parts[1]);
+            return (majorVer > major) || (majorVer == major && minorVer >= minor);
+        }
+        return false;
+    }
+
+    public boolean atLeastMySQL84() {
+        return atLeast(8, 4);
+    }
+
     private void accumulateReceivedBytes(long x) {
         if (receivedBinlogBytes != null) {
             receivedBinlogBytes.addAndGet(x);
@@ -627,7 +653,7 @@ public boolean isMixed() {
 
         private String value;
 
-        private BinlogFormat(String value){
+        private BinlogFormat(String value) {
             this.value = value;
         }
 
@@ -667,7 +693,7 @@ public boolean isNoBlob() {
 
         private String value;
 
-        private BinlogImage(String value){
+        private BinlogImage(String value) {
             this.value = value;
         }
 

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java

@@ -668,13 +668,20 @@ private Long findServerId(MysqlConnection mysqlConnection) {
      * 查询当前的binlog位置
      */
     private EntryPosition findEndPosition(MysqlConnection mysqlConnection) {
+        String showSql = "show master status";
         try {
-            String showSql = multiStreamEnable ? "show master status with " + destination : "show master status";
+            if (mysqlConnection.atLeastMySQL84()) {
+                // 8.4新语法
+                showSql = "show binary log status";
+            } else if (multiStreamEnable) {
+                // 兼容polardb-x的多流binlog
+                showSql = "show master status with " + destination;
+            }
             ResultSetPacket packet = mysqlConnection.query(showSql);
             List<String> fields = packet.getFieldValues();
             if (CollectionUtils.isEmpty(fields)) {
                 throw new CanalParseException(
-                        "command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
+                    "command : '" + showSql + "' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
             }
             EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
             if (isGTIDMode() && fields.size() > 4) {
@@ -690,7 +697,7 @@ private EntryPosition findEndPosition(MysqlConnection mysqlConnection) {
             }
             return endPosition;
         } catch (IOException e) {
-            throw new CanalParseException("command : 'show master status' has an error!", e);
+            throw new CanalParseException("command : '" + showSql + "' has an error!", e);
         }
     }
 
@@ -721,7 +728,12 @@ private EntryPosition findStartPosition(MysqlConnection mysqlConnection) {
     @SuppressWarnings("unused")
     private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) {
         try {
-            ResultSetPacket packet = mysqlConnection.query("show slave status");
+            String showSql = "show slave stauts";
+            if (mysqlConnection.atLeastMySQL84()) {
+                // 兼容mysql 8.4
+                showSql = "show replica status";
+            }
+            ResultSetPacket packet = mysqlConnection.query(showSql);
             List<FieldPacket> names = packet.getFieldDescriptors();
             List<String> fields = packet.getFieldValues();
             if (CollectionUtils.isEmpty(fields)) {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java

@@ -195,7 +195,7 @@ private String makeRequestString(Map<String, String> param) throws Exception {
     private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception {
         SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (TrustStrategy) (arg0, arg1) -> true).build();
         SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext,
-            new String[] { "TLSv1" },
+            new String[] { "TLSv1.2" },
             null,
             SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
         Registry registry = RegistryBuilder.create()

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java

@@ -207,6 +207,11 @@ private boolean dumpTableMeta(MysqlConnection connection, final CanalEventFilter
             }
 
             for (String schema : schemas) {
+                // 如果schema命中黑名单，直接跳过
+                // 解决部分数据库可以看到database，但实际表级别无权限的情况
+                if (blackFilter != null && blackFilter.filter(schema + ".%")) {
+                    continue;
+                }
                 // filter views
                 packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
                 columnSize = packet.getFieldDescriptors().size();