diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml index d06eed6f..3824e012 100644 --- a/client-adapter/elasticsearch/pom.xml +++ b/client-adapter/elasticsearch/pom.xml @@ -21,7 +21,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_644 + 2.0.0_preview_855 org.elasticsearch diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 1c505cc9..1652d351 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -17,6 +17,7 @@ import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.transport.client.PreBuiltTransportClient; +import org.slf4j.MDC; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; @@ -29,7 +30,12 @@ import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor; import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService; import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; -import com.alibaba.otter.canal.client.adapter.support.*; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.SPI; + /** * ES外部适配器 diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java index d4c7680f..e0c23e7b 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java @@ -9,8 +9,16 @@ import java.util.stream.Collectors; import com.alibaba.fastsql.sql.SQLUtils; import com.alibaba.fastsql.sql.ast.SQLExpr; -import com.alibaba.fastsql.sql.ast.expr.*; -import com.alibaba.fastsql.sql.ast.statement.*; +import com.alibaba.fastsql.sql.ast.expr.SQLBinaryOpExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLCaseExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr; +import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLJoinTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLSelectStatement; +import com.alibaba.fastsql.sql.ast.statement.SQLSubqueryTableSource; +import com.alibaba.fastsql.sql.ast.statement.SQLTableSource; import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; import com.alibaba.fastsql.sql.parser.ParserException; diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java index 45b0e60e..2e77d5fc 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java @@ -1,7 +1,11 @@ package com.alibaba.otter.canal.client.adapter.es.service; import java.sql.SQLException; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index ace00cba..d0db8606 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -7,7 +7,6 @@ import java.util.Map; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.support.Util; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -23,6 +22,7 @@ import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.Util; /** * ES 同步 Service diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java index b49e3fcc..52baf3c3 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java @@ -5,7 +5,13 @@ import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.sql.Blob; import java.sql.SQLException; -import java.util.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; import org.apache.commons.codec.binary.Base64; import org.joda.time.DateTime; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java index 966b115b..7438a662 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java index e8cbbe59..4b28626e 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java index 0fba73d3..d403a6eb 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java index 9739a355..ec110c81 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java index c0e43b61..f88b858e 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import javax.sql.DataSource; diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java index 6bf4b75b..90e23cea 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import org.elasticsearch.action.get.GetResponse; import org.junit.Assert; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 11c26f4a..9974928d 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -2,7 +2,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.net.InetSocketAddress; import java.net.SocketAddress; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java index 2045b822..32b39f47 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java @@ -1,13 +1,23 @@ package com.alibaba.otter.canal.adapter.launcher.rest; -import java.util.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; import javax.annotation.PostConstruct; import javax.annotation.Resource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.web.bind.annotation.*; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.PutMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; import com.alibaba.otter.canal.adapter.launcher.common.EtlLock; import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index ed5f44e2..25add346 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -1,6 +1,10 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; -import java.sql.*; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index da5a0479..25247642 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -4,7 +4,14 @@ import java.io.Reader; import java.io.StringReader; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; -import java.sql.*; +import java.sql.Blob; +import java.sql.Clob; +import java.sql.Date; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Time; +import java.sql.Timestamp; +import java.sql.Types; import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; diff --git a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java index 5ea374e2..76ae1c21 100644 --- a/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java +++ b/client-adapter/rdb/src/test/java/com/alibaba/otter/canal/client/adapter/rdb/test/sync/OracleSyncTest.java @@ -1,6 +1,11 @@ package com.alibaba.otter.canal.client.adapter.rdb.test.sync; -import java.util.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import org.junit.Before; import org.junit.Test; diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index bc1db836..514a231f 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -33,7 +33,6 @@ canal.instance.tsdb.enable=true canal.instance.dbUsername=canal canal.instance.dbPassword=canal canal.instance.connectionCharset = UTF-8 -canal.instance.defaultDatabaseName =test # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 019c3031..164546d9 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -136,7 +136,7 @@ - + @@ -146,7 +146,7 @@ - + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 636dc4e5..87fdc49f 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -121,7 +121,7 @@ - + @@ -131,7 +131,7 @@ - + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index d0c94057..9f96801e 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -118,7 +118,7 @@ - + @@ -128,7 +128,7 @@ - + @@ -220,7 +220,7 @@ - + @@ -230,7 +230,7 @@ - + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 68f90f0c..d7289aa4 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -109,7 +109,7 @@ - + @@ -119,7 +119,7 @@ - + diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java index 0f01cbd5..ed5e6526 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java @@ -2,19 +2,24 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; import java.net.InetSocketAddress; +import java.security.NoSuchAlgorithmException; import java.util.concurrent.atomic.AtomicBoolean; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.client.AuthSwitchResponsePacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.AuthSwitchRequestMoreData; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.AuthSwitchRequestPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.HandshakeInitializationPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.Reply323Packet; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool; +import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC; import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; @@ -32,7 +37,7 @@ public class MysqlConnector { private String password; private byte charsetNumber = 33; - private String defaultSchema = "test"; + private String defaultSchema; private int soTimeout = 30 * 1000; private int connTimeout = 5 * 1000; private int receiveBufferSize = 16 * 1024; @@ -150,6 +155,7 @@ public class MysqlConnector { } private void negotiate(SocketChannel channel) throws IOException { + // https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol HeaderPacket header = PacketManager.readHeader(channel, 4, timeout); byte[] body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); if (body[0] < 0) {// check field_count @@ -165,10 +171,14 @@ public class MysqlConnector { } HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket(); handshakePacket.fromBytes(body); + if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) { + // HandshakeV9 + auth323(channel, (byte) (header.getPacketSequenceNumber() + 1), handshakePacket.seed); + return; + } + connectionId = handshakePacket.threadId; // 记录一下connection - logger.info("handshake initialization packet received, prepare the client authentication packet to send"); - ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket(); clientAuth.setCharsetNumber(charsetNumber); @@ -177,6 +187,7 @@ public class MysqlConnector { clientAuth.setServerCapabilities(handshakePacket.serverCapabilities); clientAuth.setDatabaseName(defaultSchema); clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket)); + clientAuth.setAuthPluginName("mysql_native_password".getBytes()); byte[] clientAuthPkgBody = clientAuth.toBytes(); HeaderPacket h = new HeaderPacket(); @@ -192,15 +203,51 @@ public class MysqlConnector { body = null; body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); assert body != null; + byte marker = body[0]; + if (marker == -2 || marker == 1) { + byte[] authData = null; + String pluginName = null; + if (marker == 1) { + AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData(); + packet.fromBytes(body); + authData = packet.authData; + } else { + AuthSwitchRequestPacket packet = new AuthSwitchRequestPacket(); + packet.fromBytes(body); + authData = packet.authData; + pluginName = packet.authName; + } + + if (pluginName != null && "mysql_native_password".equals(pluginName)) { + byte[] encryptedPassword = null; + try { + encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData); + } catch (NoSuchAlgorithmException e) { + throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e); + } + AuthSwitchResponsePacket responsePacket = new AuthSwitchResponsePacket(); + responsePacket.authData = encryptedPassword; + byte[] auth = responsePacket.toBytes(); + + h = new HeaderPacket(); + h.setPacketBodyLength(auth.length); + h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1)); + PacketManager.writePkg(channel, h.toBytes(), auth); + logger.info("auth switch response packet is sent out."); + + header = null; + header = PacketManager.readHeader(channel, 4); + body = null; + body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); + assert body != null; + } + } + if (body[0] < 0) { if (body[0] == -1) { ErrorPacket err = new ErrorPacket(); err.fromBytes(body); throw new IOException("Error When doing Client Authentication:" + err.toString()); - } else if (body[0] == -2) { - auth323(channel, header.getPacketSequenceNumber(), handshakePacket.seed); - // throw new - // IOException("Unexpected EOF packet at Client Authentication."); } else { throw new IOException("unpexpected packet with field_count=" + body[0]); } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/Capability.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/Capability.java new file mode 100644 index 00000000..20428f36 --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/Capability.java @@ -0,0 +1,170 @@ +package com.alibaba.otter.canal.parse.driver.mysql.packets; + +/** + * https://dev.mysql.com/doc/internals/en/capability-flags.html#packet-Protocol + * ::CapabilityFlags + */ +public interface Capability { + + // Use the improved version of Old Password Authentication. + // Assumed to be set since 4.1.1. + int CLIENT_LONG_PASSWORD = 0x00000001; + + // Send found rows instead of affected rows in EOF_Packet. + int CLIENT_FOUND_ROWS = 0x00000002; + + // https://dev.mysql.com/doc/internals/en/com-query-response.html#packet-Protocol::ColumnDefinition320 + // Longer flags in Protocol::ColumnDefinition320. + // Server:Supports longer flags. + // Client:Expects longer flags. + // 执行查询sql时，除了返回结果集，还返回元数据 + int CLIENT_LONG_FLAG = 0x00000004; + + // 可以在handshake时，指定一个数据库名 + // Database (schema) name can be specified on connect in Handshake Response + // Packet. + // Server: Supports schema-name in Handshake Response Packet. + // Client: Handshake Response Packet contains a schema-name. + int CLIENT_CONNECT_WITH_DB = 0x00000008; + + // Server: Do not permit database.table.column. + int CLIENT_NO_SCHEMA = 0x00000010; + + // Compression protocol supported. + // Server:Supports compression. + // Client:Switches to Compression compressed protocol after successful + // authentication. + int CLIENT_COMPRESS = 0x00000020; + + // Special handling of ODBC behavior. + // No special behavior since 3.22. + int CLIENT_ODBC = 0x00000040; + + // Can use LOAD DATA LOCAL. + // Server:Enables the LOCAL INFILE request of LOAD DATA|XML. + // Client:Will handle LOCAL INFILE request. + int CLIENT_LOCAL_FILES = 0x00000080; + + // Server: Parser can ignore spaces before '('. + // Client: Let the parser ignore spaces before '('. + int CLIENT_IGNORE_SPACE = 0x00000100; + + // Server:Supports the 4.1 protocol, + // 4.1协议中， + // OKPacket将会包含warning count + // ERR_Packet包含SQL state + // EOF_Packet包含warning count和status flags + // Client:Uses the 4.1 protocol. + // Note: this value was CLIENT_CHANGE_USER in 3.22, unused in 4.0 + // If CLIENT_PROTOCOL_41 is set： + // 1、the ok packet contains a warning count. + // https://dev.mysql.com/doc/internals/en/packet-OK_Packet.html + // 2、ERR_Packet It contains a SQL state value if CLIENT_PROTOCOL_41 is + // enabled. //https://dev.mysql.com/doc/internals/en/packet-ERR_Packet.html + // 3、EOF_Packet If CLIENT_PROTOCOL_41 is enabled, the EOF packet contains a + // warning count and status flags. + // https://dev.mysql.com/doc/internals/en/packet-EOF_Packet.html + int CLIENT_PROTOCOL_41 = 0x00000200; + + // wait_timeout versus wait_interactive_timeout. + // Server:Supports interactive and noninteractive clients. + // Client:Client is interactive. + int CLIENT_INTERACTIVE = 0x00000400; + + // Server: Supports SSL. + // Client: Switch to SSL after sending the capability-flags. + int CLIENT_SSL = 0x00000800; + + // Client: Do not issue SIGPIPE if network failures occur (libmysqlclient + // only). + int CLIENT_IGNORE_SIGPIPE = 0x00001000; + + // Server: Can send status flags in EOF_Packet. + // Client:Expects status flags in EOF_Packet. + // Note:This flag is optional in 3.23, but always set by the server since + // 4.0. + int CLIENT_TRANSACTIONS = 0x00002000; + + // Unused + // Note: Was named CLIENT_PROTOCOL_41 in 4.1.0. + int CLIENT_RESERVED = 0x00004000; + + /** + * 
+     *      服务端返回20 byte随机字节，客户端利用其对密码进行加密，加密算法如下：
+     *      https://dev.mysql.com/doc/internals/en/secure-password-authentication.html#packet-Authentication::Native41
+     *      Authentication::Native41:
+     *      client-side expects a 20-byte random challenge
+     *      client-side returns a 20-byte response based on the algorithm described later
+     *      Name
+     *      mysql_native_password
+     *      Requires
+     *      CLIENT_SECURE_CONNECTION
+     *      Image description follows.
+     *      Image description
+     *      This method fixes a 2 short-comings of the Old Password Authentication:
+     *      (https://dev.mysql.com/doc/internals/en/old-password-authentication.html#packet-Authentication::Old)
+     *      using a tested, crypto-graphic hashing function which isn't broken
+     *      knowning the content of the hash in the mysql.user table isn't enough to authenticate against the MySQL Server.
+     *      The password is calculated by:
+     *      SHA1( password ) XOR SHA1( "20-bytes random data from server"  SHA1( SHA1( password ) ) )
+     *
+ */ + int CLIENT_SECURE_CONNECTION = 0x00008000; + + // Server:Can handle multiple statements per COM_QUERY and COM_STMT_PREPARE. + // Client:May send multiple statements per COM_QUERY and COM_STMT_PREPARE. + // Note:Was named CLIENT_MULTI_QUERIES in 4.1.0, renamed later. + // Requires:CLIENT_PROTOCOL_41 + int CLIENT_MULTI_STATEMENTS = 0x00010000; + + // Server: Can send multiple resultsets for COM_QUERY. + // Client: Can handle multiple resultsets for COM_QUERY. + // Requires:CLIENT_PROTOCOL_41 + int CLIENT_MULTI_RESULTS = 0x00020000; + + // Server: Can send multiple resultsets for ComStmtExecutePacket. + // Client: Can handle multiple resultsets for ComStmtExecutePacket. + // Requires:CLIENT_PROTOCOL_41 + int CLIENT_PS_MULTI_RESULTS = 0x00040000; + + // Server:Sends extra data in Initial Handshake Packet and supports the + // pluggable authentication protocol. + // Client: Supports authentication plugins. + // Requires: CLIENT_PROTOCOL_41 + int CLIENT_PLUGIN_AUTH = 0x00080000; + + // Server: Permits connection attributes in Protocol::HandshakeResponse41. + // Client: Sends connection attributes in Protocol::HandshakeResponse41. + int CLIENT_CONNECT_ATTRS = 0x00100000; + + // Server:Understands length-encoded integer for auth response data in + // Protocol::HandshakeResponse41. + // Client:Length of auth response data in Protocol::HandshakeResponse41 is a + // length-encoded integer. + // Note: The flag was introduced in 5.6.6, but had the wrong value. + int CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA = 0x00200000; + + // Server: Announces support for expired password extension. + // Client: Can handle expired passwords. + int CLIENT_CAN_HANDLE_EXPIRED_PASSWORDS = 0x00400000; + + // Server: Can set SERVER_SESSION_STATE_CHANGED in the Status Flags and send + // session-state change data after a OK packet. + // Client: Expects the server to send sesson-state changes after a OK + // packet. + int CLIENT_SESSION_TRACK = 0x00800000; + + /** + * Server: Can send OK after a Text Resultset. Client: Expects an OK + * (instead of EOF) after the resultset rows of a Text Resultset. + * Background:To support CLIENT_SESSION_TRACK, additional information must + * be sent after all successful commands. Although the OK packet is + * extensible, the EOF packet is not due to the overlap of its bytes with + * the content of the Text Resultset Row. Therefore, the EOF packet in the + * Text Resultset is replaced with an OK packet. EOF packets are deprecated + * as of MySQL 5.7.5. + */ + int CLIENT_DEPRECATE_EOF = 0x01000000; + +} diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/AuthSwitchResponsePacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/AuthSwitchResponsePacket.java new file mode 100644 index 00000000..5bf22374 --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/AuthSwitchResponsePacket.java @@ -0,0 +1,21 @@ +package com.alibaba.otter.canal.parse.driver.mysql.packets.client; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket; + +public class AuthSwitchResponsePacket extends CommandPacket { + + public byte[] authData; + + public void fromBytes(byte[] data) { + } + + public byte[] toBytes() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write(authData); + return out.toByteArray(); + } + +} diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java index fb04d7dc..1750b255 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java @@ -6,6 +6,7 @@ import java.security.NoSuchAlgorithmException; import org.apache.commons.lang.StringUtils; +import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability; import com.alibaba.otter.canal.parse.driver.mysql.packets.PacketWithHeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC; @@ -13,12 +14,17 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter; public class ClientAuthenticationPacket extends PacketWithHeaderPacket { + private int clientCapability = Capability.CLIENT_LONG_PASSWORD | Capability.CLIENT_LONG_FLAG + | Capability.CLIENT_PROTOCOL_41 | Capability.CLIENT_INTERACTIVE + | Capability.CLIENT_TRANSACTIONS | Capability.CLIENT_SECURE_CONNECTION + | Capability.CLIENT_MULTI_STATEMENTS | Capability.CLIENT_PLUGIN_AUTH; private String username; private String password; private byte charsetNumber; private String databaseName; private int serverCapabilities; private byte[] scrumbleBuff; + private byte[] authPluginName; public void fromBytes(byte[] data) { // bypass since nowhere to use. @@ -36,6 +42,7 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { * n (Null-Terminated String) user * n (Length Coded Binary) scramble_buff (1 + x bytes) * n (Null-Terminated String) databasename (optional) + * n (Null-Terminated String) auth plugin name (optional) * * * @throws IOException @@ -43,13 +50,7 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { public byte[] toBytes() throws IOException { ByteArrayOutputStream out = new ByteArrayOutputStream(); // 1. write client_flags - // 1|4|512|1024|8192|32768 - /** - * CLIENT_LONG_PASSWORD CLIENT_LONG_FLAG CLIENT_PROTOCOL_41 - * CLIENT_INTERACTIVE CLIENT_TRANSACTIONS CLIENT_SECURE_CONNECTION - * CLIENT_MULTI_STATEMENTS; - */ - ByteHelper.writeUnsignedIntLittleEndian(1 | 4 | 512 | 8192 | 32768 | 0x00010000, out); // remove + ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out); // remove // client_interactive // feature @@ -76,6 +77,10 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { if (getDatabaseName() != null) { ByteHelper.writeNullTerminatedString(getDatabaseName(), out); } + // 8 . (Null-Terminated String) auth plugin name (optional) + if (getAuthPluginName() != null) { + ByteHelper.writeNullTerminated(getAuthPluginName(), out); + } // end write return out.toByteArray(); } @@ -106,6 +111,9 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { public void setDatabaseName(String databaseName) { this.databaseName = databaseName; + if (databaseName != null) { + this.clientCapability |= Capability.CLIENT_CONNECT_WITH_DB; + } } public String getDatabaseName() { @@ -128,4 +136,15 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { return scrumbleBuff; } + public byte[] getAuthPluginName() { + return authPluginName; + } + + public void setAuthPluginName(byte[] authPluginName) { + this.authPluginName = authPluginName; + if (authPluginName != null) { + this.clientCapability |= Capability.CLIENT_PLUGIN_AUTH; + } + } + } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestMoreData.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestMoreData.java new file mode 100644 index 00000000..347f26dc --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestMoreData.java @@ -0,0 +1,25 @@ +package com.alibaba.otter.canal.parse.driver.mysql.packets.server; + +import java.io.IOException; + +import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; + +public class AuthSwitchRequestMoreData extends CommandPacket { + + public int status; + public byte[] authData; + + public void fromBytes(byte[] data) { + int index = 0; + // 1. read status + status = data[index]; + index += 1; + authData = ByteHelper.readNullTerminatedBytes(data, index); + } + + public byte[] toBytes() throws IOException { + return null; + } + +} diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestPacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestPacket.java new file mode 100644 index 00000000..3cc7de97 --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestPacket.java @@ -0,0 +1,29 @@ +package com.alibaba.otter.canal.parse.driver.mysql.packets.server; + +import java.io.IOException; + +import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; + +public class AuthSwitchRequestPacket extends CommandPacket { + + public int status; + public String authName; + public byte[] authData; + + public void fromBytes(byte[] data) { + int index = 0; + // 1. read status + status = data[index]; + index += 1; + byte[] authName = ByteHelper.readNullTerminatedBytes(data, index); + this.authName = new String(authName); + index += authName.length + 1; + authData = ByteHelper.readNullTerminatedBytes(data, index); + } + + public byte[] toBytes() throws IOException { + return null; + } + +} diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/HandshakeInitializationPacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/HandshakeInitializationPacket.java index 05a83cfd..488490ff 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/HandshakeInitializationPacket.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/HandshakeInitializationPacket.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets.server; import java.io.IOException; +import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.PacketWithHeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; @@ -23,6 +24,7 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket { public byte serverCharsetNumber; public int serverStatus; public byte[] restOfScrambleBuff; + public byte[] authPluginName; public HandshakeInitializationPacket(){ } @@ -66,20 +68,40 @@ public class HandshakeInitializationPacket extends PacketWithHeaderPacket { // 5. read server_capabilities this.serverCapabilities = ByteHelper.readUnsignedShortLittleEndian(data, index); index += 2; - // 6. read server_language - this.serverCharsetNumber = data[index]; - index++; - // 7. read server_status - this.serverStatus = ByteHelper.readUnsignedShortLittleEndian(data, index); - index += 2; - // 8. bypass filtered bytes - index += 13; - // 9. read rest of scramble_buff - this.restOfScrambleBuff = ByteHelper.readFixedLengthBytes(data, index, 12); // 虽然Handshake - // Initialization - // Packet规定最后13个byte是剩下的scrumble, - // 但实际上最后一个字节是0, 不应该包含在scrumble中. - // end read + if (data.length > index) { + // 6. read server_language + this.serverCharsetNumber = data[index]; + index++; + // 7. read server_status + this.serverStatus = ByteHelper.readUnsignedShortLittleEndian(data, index); + index += 2; + // 8. bypass filtered bytes + int capabilityFlags2 = ByteHelper.readUnsignedShortLittleEndian(data, index); + index += 2; + int capabilities = (capabilityFlags2 << 16) | this.serverCapabilities; + // int authPluginDataLen = -1; + // if ((capabilities & Capability.CLIENT_PLUGIN_AUTH) != 0) { + // authPluginDataLen = data[index]; + // } + index += 1; + index += 10; + // 9. read rest of scramble_buff + if ((capabilities & Capability.CLIENT_SECURE_CONNECTION) != 0) { + // int len = Math.max(13, authPluginDataLen - 8); + // this.authPluginDataPart2 = + // buffer.readFixedLengthString(len);// scramble2 + + // Packet规定最后13个byte是剩下的scrumble, + // 但实际上最后一个字节是0, 不应该包含在scrumble中. + this.restOfScrambleBuff = ByteHelper.readFixedLengthBytes(data, index, 12); + } + + index += 12 + 1; + if ((capabilities & Capability.CLIENT_PLUGIN_AUTH) != 0) { + this.authPluginName = ByteHelper.readNullTerminatedBytes(data, index); + } + // end read + } } /** diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java index 7841205c..a6d048ee 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.parse.driver.mysql.utils; +import java.io.ByteArrayOutputStream; import java.io.IOException; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; @@ -28,7 +29,11 @@ public abstract class PacketManager { } public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException { - ch.write(srcs); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + for (byte[] src : srcs) { + out.write(src); + } + ch.write(out.toByteArray()); } public static void writeBody(SocketChannel ch, byte[] body) throws IOException { diff --git a/instance/spring/src/test/resources/spring/default-instance.xml b/instance/spring/src/test/resources/spring/default-instance.xml index 417934da..84a50e3d 100644 --- a/instance/spring/src/test/resources/spring/default-instance.xml +++ b/instance/spring/src/test/resources/spring/default-instance.xml @@ -140,7 +140,7 @@ - + @@ -148,7 +148,7 @@ - + diff --git a/instance/spring/src/test/resources/spring/file-instance.xml b/instance/spring/src/test/resources/spring/file-instance.xml index fcd0a78f..aa717ca3 100644 --- a/instance/spring/src/test/resources/spring/file-instance.xml +++ b/instance/spring/src/test/resources/spring/file-instance.xml @@ -128,7 +128,7 @@ - + @@ -136,7 +136,7 @@ - + diff --git a/instance/spring/src/test/resources/spring/group-instance.xml b/instance/spring/src/test/resources/spring/group-instance.xml index 763ec12f..4dea968a 100644 --- a/instance/spring/src/test/resources/spring/group-instance.xml +++ b/instance/spring/src/test/resources/spring/group-instance.xml @@ -123,7 +123,7 @@ - + @@ -131,7 +131,7 @@ - + @@ -203,7 +203,7 @@ - + @@ -211,7 +211,7 @@ - + diff --git a/instance/spring/src/test/resources/spring/memory-instance.xml b/instance/spring/src/test/resources/spring/memory-instance.xml index 0f58e964..f9e98475 100644 --- a/instance/spring/src/test/resources/spring/memory-instance.xml +++ b/instance/spring/src/test/resources/spring/memory-instance.xml @@ -114,7 +114,7 @@ - + @@ -122,7 +122,7 @@ - + diff --git a/pom.xml b/pom.xml index 61d91a28..567d6bb9 100644 --- a/pom.xml +++ b/pom.xml @@ -246,7 +246,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_836 + 2.0.0_preview_855 com.alibaba diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index 17568d44..fc27baca 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -36,8 +36,11 @@ public class MQMessageUtils { public List apply(String pkHashConfigs) { List datas = Lists.newArrayList(); - String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, - ","); + + String[] pkHashConfigArray = StringUtils.split(StringUtils.replace(pkHashConfigs, + ",", + ";"), + ";"); // schema.table:id^name for (String pkHashConfig : pkHashConfigArray) { PartitionData data = new PartitionData(); @@ -75,8 +78,10 @@ public class MQMessageUtils { public List apply(String pkHashConfigs) { List datas = Lists.newArrayList(); - String[] dynamicTopicArray = StringUtils.split(pkHashConfigs, - ","); + String[] dynamicTopicArray = StringUtils.split(StringUtils.replace(pkHashConfigs, + ",", + ";"), + ";"); // schema.table for (String dynamicTopic : dynamicTopicArray) { DynamicTopicData data = new DynamicTopicData(); @@ -216,7 +221,7 @@ public class MQMessageUtils { partitionEntries[pkHash].add(entry); } else { for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - int hashCode = table.hashCode(); + int hashCode = database.hashCode(); if (hashMode.autoPkHash) { // isEmpty use default pkNames for (CanalEntry.Column column : rowData.getAfterColumnsList()) { @@ -432,7 +437,7 @@ public class MQMessageUtils { int idx = 0; for (Map row : flatMessage.getData()) { - int hashCode = table.hashCode(); + int hashCode = database.hashCode(); for (String pkName : pkNames) { String value = row.get(pkName); if (value == null) { @@ -509,7 +514,7 @@ public class MQMessageUtils { } private static Set matchTopics(String name, String dynamicTopicConfigs) { - String[] router = StringUtils.split(dynamicTopicConfigs, ','); + String[] router = StringUtils.split(StringUtils.replace(dynamicTopicConfigs, ",", ";"), ";"); Set topics = new HashSet<>(); for (String item : router) { int i = item.indexOf(":"); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index a3a8aec5..6cc6724f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.kafka; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Properties; @@ -138,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer { private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message) throws Exception { if (!kafkaProperties.getFlatMessage()) { - ProducerRecord record = null; + List> records = new ArrayList>(); if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { Message[] messages = MQMessageUtils.messagePartition(message, canalDestination.getPartitionsNum(), @@ -147,18 +148,16 @@ public class CanalKafkaProducer implements CanalMQProducer { for (int i = 0; i < length; i++) { Message messagePartition = messages[i]; if (messagePartition != null) { - record = new ProducerRecord<>(topicName, i, null, messagePartition); + records.add(new ProducerRecord(topicName, i, null, messagePartition)); } } } else { final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0; - record = new ProducerRecord<>(topicName, partition, null, message); + records.add(new ProducerRecord(topicName, partition, null, message)); } - if (record != null) { - if (kafkaProperties.getTransaction()) { - producer.send(record).get(); - } else { + if (!records.isEmpty()) { + for (ProducerRecord record : records) { producer.send(record).get(); } diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index fa1a2819..aaa658e2 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -107,11 +107,10 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public MessageQueue select(List mqs, Message msg, Object arg) { if (index > mqs.size()) { - throw new CanalServerException("partition number is error,config num:" - + destination.getPartitionsNum() - + ", mq num: " + mqs.size()); + return mqs.get(index % mqs.size()); + } else { + return mqs.get(index); } - return mqs.get(index); } }, null); } catch (Exception e) { @@ -134,7 +133,11 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public MessageQueue select(List mqs, Message msg, Object arg) { - return mqs.get(partition); + if (partition > mqs.size()) { + return mqs.get(partition % mqs.size()); + } else { + return mqs.get(partition); + } } }, null); } @@ -168,11 +171,10 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public MessageQueue select(List mqs, Message msg, Object arg) { if (index > mqs.size()) { - throw new CanalServerException("partition number is error,config num:" - + destination.getPartitionsNum() - + ", mq num: " + mqs.size()); + return mqs.get(index % mqs.size()); + } else { + return mqs.get(index); } - return mqs.get(index); } }, null); } catch (Exception e) { @@ -196,7 +198,11 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public MessageQueue select(List mqs, Message msg, Object arg) { - return mqs.get(partition); + if (partition > mqs.size()) { + return mqs.get(partition % mqs.size()); + } else { + return mqs.get(partition); + } } }, null); } catch (Exception e) {