From 0ec46991499a22870dde4ae736b2586cbcbfea94 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 27 Mar 2019 11:54:40 +0800 Subject: [PATCH 01/11] fixed format & support Mysql Auth Switch Protocol --- .../canal/client/adapter/es/ESAdapter.java | 8 +- .../client/adapter/es/config/SqlParser.java | 12 +- .../adapter/es/service/ESEtlService.java | 6 +- .../adapter/es/service/ESSyncService.java | 2 +- .../client/adapter/es/support/ESSyncUtil.java | 8 +- .../es/test/sync/LabelSyncJoinSub2Test.java | 6 +- .../es/test/sync/LabelSyncJoinSubTest.java | 6 +- .../es/test/sync/RoleSyncJoinOne2Test.java | 6 +- .../es/test/sync/RoleSyncJoinOneTest.java | 6 +- .../es/test/sync/UserSyncJoinOneTest.java | 6 +- .../es/test/sync/UserSyncSingleTest.java | 6 +- .../launcher/loader/CanalAdapterLoader.java | 6 +- .../adapter/launcher/rest/CommonRest.java | 14 +- .../adapter/rdb/service/RdbEtlService.java | 6 +- .../client/adapter/rdb/support/SyncUtil.java | 9 +- .../adapter/rdb/test/sync/OracleSyncTest.java | 7 +- .../parse/driver/mysql/MysqlConnector.java | 59 +++++- .../driver/mysql/packets/Capability.java | 170 ++++++++++++++++++ .../client/AuthSwitchResponsePacket.java | 21 +++ .../client/ClientAuthenticationPacket.java | 33 +++- .../server/AuthSwitchRequestMoreData.java | 25 +++ .../server/AuthSwitchRequestPacket.java | 29 +++ .../server/HandshakeInitializationPacket.java | 50 ++++-- .../driver/mysql/MysqlConnectorTest.java | 33 +--- .../otter/canal/common/MQMessageUtils.java | 15 +- 25 files changed, 469 insertions(+), 80 deletions(-) create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/Capability.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/AuthSwitchResponsePacket.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestMoreData.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/AuthSwitchRequestPacket.java 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 b41b857b..68962a2a 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,8 +30,11 @@ 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 org.slf4j.MDC; +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/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..3a4c1f4c 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; @@ -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(handshakePacket.authPluginName); 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..5a119905 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; 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/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java index abcd1473..1af5d1e3 100644 --- a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java +++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java @@ -3,51 +3,26 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; import java.net.InetSocketAddress; -import org.junit.Assert; -import org.junit.Test; - import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; public class MysqlConnectorTest { - @Test - public void testQuery() { - - MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx"); + public static void main(String args[]) { + MysqlConnector connector = new MysqlConnector(new InetSocketAddress("yddb01.mysql.database.chinacloudapi.cn", + 3306), "ps-admin01@yddb01", "1qaz3edc"); try { connector.connect(); MysqlQueryExecutor executor = new MysqlQueryExecutor(connector); ResultSetPacket result = executor.query("show variables like '%char%';"); System.out.println(result); - result = executor.query("select * from test.test1"); - System.out.println(result); } catch (IOException e) { - Assert.fail(e.getMessage()); + e.printStackTrace(); } finally { try { connector.disconnect(); } catch (IOException e) { - Assert.fail(e.getMessage()); } } } - // @Test - public void testUpdate() { - - MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx"); - try { - connector.connect(); - MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector); - executor.update("insert into test.test2(id,name,score,text_value) values(null,'中文1',10,'中文2')"); - } catch (IOException e) { - Assert.fail(e.getMessage()); - } finally { - try { - connector.disconnect(); - } catch (IOException e) { - Assert.fail(e.getMessage()); - } - } - } } 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..acff1751 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(); @@ -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(":"); From 739963acd03c9633416ccca85a80ef1cc8a6ff40 Mon Sep 17 00:00:00 2001 From: wangsaner Date: Wed, 27 Mar 2019 21:57:04 +0800 Subject: [PATCH 02/11] =?UTF-8?q?fixed=20issue=20#1636=20=E9=9D=9Eflat?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F=EF=BC=8C=E5=BD=93MQ=E4=B8=BB=E9=A2=98?= =?UTF-8?q?=E4=B8=BA=E5=A4=9A=E5=88=86=E5=8C=BA=E6=97=B6=EF=BC=8C=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E6=8B=86=E5=88=86=E5=8F=AA=E5=8F=96=E4=BA=86for?= =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E9=87=8C=E9=9D=A2=E7=9A=84=E6=9C=80=E5=90=8E?= =?UTF-8?q?=E4=B8=80=E4=BB=BD=E6=95=B0=E6=8D=AE=EF=BC=8C=20=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E6=B6=88=E8=B4=B9=E7=AB=AF=E5=8F=AA=E8=83=BD=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=E5=88=B0=E4=B8=80=E4=B8=AA=E5=88=86=E5=8C=BA=E7=9A=84?= =?UTF-8?q?=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 27 +++++++++---------- 1 file changed, 13 insertions(+), 14 deletions(-) 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..c4605067 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,25 +148,23 @@ 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 { - producer.send(record).get(); - } - - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); - } - } + if (!records.isEmpty()) { + for (ProducerRecord record : records) { + producer.send(record).get(); + } + + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); + } + } } else { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message); From 20e7becfe90b62e8f95bdaf870a8e4de5c8dc48c Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 10:19:53 +0800 Subject: [PATCH 03/11] fixed format --- .../driver/mysql/MysqlConnectorTest.java | 33 ++++++++++++++++--- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java index 1af5d1e3..abcd1473 100644 --- a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java +++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java @@ -3,26 +3,51 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; import java.net.InetSocketAddress; +import org.junit.Assert; +import org.junit.Test; + import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; public class MysqlConnectorTest { - public static void main(String args[]) { - MysqlConnector connector = new MysqlConnector(new InetSocketAddress("yddb01.mysql.database.chinacloudapi.cn", - 3306), "ps-admin01@yddb01", "1qaz3edc"); + @Test + public void testQuery() { + + MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx"); try { connector.connect(); MysqlQueryExecutor executor = new MysqlQueryExecutor(connector); ResultSetPacket result = executor.query("show variables like '%char%';"); System.out.println(result); + result = executor.query("select * from test.test1"); + System.out.println(result); } catch (IOException e) { - e.printStackTrace(); + Assert.fail(e.getMessage()); } finally { try { connector.disconnect(); } catch (IOException e) { + Assert.fail(e.getMessage()); } } } + // @Test + public void testUpdate() { + + MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx"); + try { + connector.connect(); + MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector); + executor.update("insert into test.test2(id,name,score,text_value) values(null,'中文1',10,'中文2')"); + } catch (IOException e) { + Assert.fail(e.getMessage()); + } finally { + try { + connector.disconnect(); + } catch (IOException e) { + Assert.fail(e.getMessage()); + } + } + } } From 97640b717b855ab8b56e3ca37026732e5d56eebd Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 10:23:33 +0800 Subject: [PATCH 04/11] fixed compiler --- .../otter/canal/kafka/CanalKafkaProducer.java | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) 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 c4605067..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 @@ -139,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer { private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message) throws Exception { if (!kafkaProperties.getFlatMessage()) { - List> records = new ArrayList<>(); + List> records = new ArrayList>(); if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { Message[] messages = MQMessageUtils.messagePartition(message, canalDestination.getPartitionsNum(), @@ -148,23 +148,23 @@ public class CanalKafkaProducer implements CanalMQProducer { for (int i = 0; i < length; i++) { Message messagePartition = messages[i]; if (messagePartition != null) { - records.add(new ProducerRecord<>(topicName, i, null, messagePartition)); + records.add(new ProducerRecord(topicName, i, null, messagePartition)); } } } else { final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0; - records.add(new ProducerRecord<>(topicName, partition, null, message)); + records.add(new ProducerRecord(topicName, partition, null, message)); } - if (!records.isEmpty()) { - for (ProducerRecord record : records) { - producer.send(record).get(); - } - - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); - } - } + if (!records.isEmpty()) { + for (ProducerRecord record : records) { + producer.send(record).get(); + } + + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); + } + } } else { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message); From 9826948ebb885dedebd6bd52fae5363e94286040 Mon Sep 17 00:00:00 2001 From: wangsaner Date: Thu, 28 Mar 2019 16:01:02 +0800 Subject: [PATCH 05/11] =?UTF-8?q?=E4=BF=AE=E6=94=B9canal.mq.partitionHash?= =?UTF-8?q?=E6=95=A3=E5=88=97=E8=A7=84=E5=88=99=EF=BC=8C=20=E5=BD=93?= =?UTF-8?q?=E6=8C=87=E6=98=8Epk=E7=BA=A7=E5=81=9A=E5=88=86=E5=8C=BA?= =?UTF-8?q?=EF=BC=88schema.table:pk1^pk2=EF=BC=89=E6=97=B6=EF=BC=8C?= =?UTF-8?q?=E7=94=B1table.hashCode()^column.getValue().hashCode()=E6=94=B9?= =?UTF-8?q?=E4=B8=BAdatabase.hashCode()^column.getValue().hashCode()?= =?UTF-8?q?=EF=BC=8C=E4=BB=8E=E8=80=8C=E6=94=AF=E6=8C=81=E5=90=8C=E5=BA=93?= =?UTF-8?q?=E4=B8=8B=E4=B8=8D=E5=90=8C=E8=A1=A8hash=E8=87=B3=E7=9B=B8?= =?UTF-8?q?=E5=90=8C=E5=88=86=E5=8C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/java/com/alibaba/otter/canal/common/MQMessageUtils.java | 1 + 1 file changed, 1 insertion(+) 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 acff1751..f1ed1c86 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 @@ -230,6 +230,7 @@ public class MQMessageUtils { } } } else { + hashCode = database.hashCode(); for (CanalEntry.Column column : rowData.getAfterColumnsList()) { if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { hashCode = hashCode ^ column.getValue().hashCode(); From a7f9a779ad26a7bb3ec614dad604f25e03cd39c3 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 17:01:36 +0800 Subject: [PATCH 06/11] fixed issue #1640 , support schema partition --- .../java/com/alibaba/otter/canal/common/MQMessageUtils.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 f1ed1c86..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 @@ -221,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()) { @@ -230,7 +230,6 @@ public class MQMessageUtils { } } } else { - hashCode = database.hashCode(); for (CanalEntry.Column column : rowData.getAfterColumnsList()) { if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { hashCode = hashCode ^ column.getValue().hashCode(); @@ -438,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) { From f63ce4ffe623a6bbcaca5023fd3512e493d442c0 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 20:45:16 +0800 Subject: [PATCH 07/11] fixed mysql auth --- deployer/src/main/resources/example/instance.properties | 1 - .../otter/canal/parse/driver/mysql/MysqlConnector.java | 4 ++-- .../mysql/packets/client/ClientAuthenticationPacket.java | 2 +- 3 files changed, 3 insertions(+), 4 deletions(-) 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/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 3a4c1f4c..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 @@ -37,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; @@ -187,7 +187,7 @@ public class MysqlConnector { clientAuth.setServerCapabilities(handshakePacket.serverCapabilities); clientAuth.setDatabaseName(defaultSchema); clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket)); - clientAuth.setAuthPluginName(handshakePacket.authPluginName); + clientAuth.setAuthPluginName("mysql_native_password".getBytes()); byte[] clientAuthPkgBody = clientAuth.toBytes(); HeaderPacket h = new HeaderPacket(); 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 5a119905..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 @@ -17,7 +17,7 @@ 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_MULTI_STATEMENTS | Capability.CLIENT_PLUGIN_AUTH; private String username; private String password; private byte charsetNumber; From d715ff64b41c90b26932233062eef016aa2c6441 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 21:52:51 +0800 Subject: [PATCH 08/11] fixed packet write batch --- .../canal/parse/driver/mysql/utils/PacketManager.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 { From 15e29d276a9dc3ca7c3204bd5d15714c1d64fa1a Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 22:05:56 +0800 Subject: [PATCH 09/11] fixed default xml config --- deployer/src/main/resources/spring/default-instance.xml | 4 ++-- deployer/src/main/resources/spring/file-instance.xml | 4 ++-- deployer/src/main/resources/spring/group-instance.xml | 8 ++++---- deployer/src/main/resources/spring/memory-instance.xml | 4 ++-- .../spring/src/test/resources/spring/default-instance.xml | 4 ++-- .../spring/src/test/resources/spring/file-instance.xml | 4 ++-- .../spring/src/test/resources/spring/group-instance.xml | 8 ++++---- .../spring/src/test/resources/spring/memory-instance.xml | 4 ++-- 8 files changed, 20 insertions(+), 20 deletions(-) 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/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 @@ - + From d3195ac9736a259dc167f343240415291bc00f17 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 3 Apr 2019 10:58:43 +0800 Subject: [PATCH 10/11] fixed issue #1626 , support alter table id first --- client-adapter/elasticsearch/pom.xml | 2 +- pom.xml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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/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 From 82ed9db8aea313421b7d6fa42bc10a84c38d8a65 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 3 Apr 2019 11:26:31 +0800 Subject: [PATCH 11/11] fixed rocketmq partition out of range --- .../canal/rocketmq/CanalRocketMQProducer.java | 26 ++++++++++++------- 1 file changed, 16 insertions(+), 10 deletions(-) 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) {