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 91d2f6f2..4850e249 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 @@ -13,11 +13,7 @@ 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.packets.server.*; 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; @@ -247,6 +243,7 @@ public class MysqlConnector { body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); assert body != null; if (body[0] == 0x01 && body[1] == 0x04) { + // fixed issue https://github.com/alibaba/canal/pull/4767, support mysql 8.0.30+ header = cachingSha2PasswordFullAuth(channel, header, getPassword().getBytes(), scramble); body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); } else { diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/MySQLPasswordEncrypter.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/MySQLPasswordEncrypter.java index 309c5b2f..b6f8a60a 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/MySQLPasswordEncrypter.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/MySQLPasswordEncrypter.java @@ -1,13 +1,9 @@ package com.alibaba.otter.canal.parse.driver.mysql.utils; -import java.security.DigestException; -import java.security.InvalidKeyException; -import java.security.KeyFactory; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; -import java.security.PublicKey; +import java.security.*; import java.security.spec.InvalidKeySpecException; import java.security.spec.X509EncodedKeySpec; + import javax.crypto.BadPaddingException; import javax.crypto.Cipher; import javax.crypto.IllegalBlockSizeException; diff --git a/example/src/main/java/com/alibaba/otter/canal/example/BaseCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/BaseCanalClientTest.java index 12ad9774..f7d40387 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/BaseCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/BaseCanalClientTest.java @@ -13,26 +13,19 @@ import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.CanalEntry; -import com.alibaba.otter.canal.protocol.CanalEntry.Column; -import com.alibaba.otter.canal.protocol.CanalEntry.Entry; -import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; -import com.alibaba.otter.canal.protocol.CanalEntry.EventType; -import com.alibaba.otter.canal.protocol.CanalEntry.Pair; -import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; -import com.alibaba.otter.canal.protocol.CanalEntry.RowData; -import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin; -import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd; import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.CanalEntry.*; import com.google.protobuf.InvalidProtocolBufferException; public class BaseCanalClientTest { - protected final static Logger logger = LoggerFactory.getLogger(AbstractCanalClientTest.class); + protected final static Logger logger = LoggerFactory + .getLogger(AbstractCanalClientTest.class); protected static final String SEP = SystemUtils.LINE_SEPARATOR; protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss"; protected volatile boolean running = false; protected Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", - e); + e); protected Thread thread = null; protected CanalConnector connector; protected static String context_format = null; @@ -51,8 +44,7 @@ public class BaseCanalClientTest { + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , gtid : ({}) , delay : {} ms" + SEP; - transaction_format = SEP - + "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms" + transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms" + SEP; } @@ -71,8 +63,8 @@ public class BaseCanalClientTest { } SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT); - logger.info(context_format, new Object[] { batchId, size, memsize, format.format(new Date()), startPosition, - endPosition }); + logger.info(context_format, + new Object[] { batchId, size, memsize, format.format(new Date()), startPosition, endPosition }); } protected String buildPositionForDump(Entry entry) { @@ -94,7 +86,8 @@ public class BaseCanalClientTest { Date date = new Date(entry.getHeader().getExecuteTime()); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { + if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN + || entry.getEntryType() == EntryType.TRANSACTIONEND) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) { TransactionBegin begin = null; try { @@ -105,9 +98,10 @@ public class BaseCanalClientTest { // 打印事务头信息，执行的线程id，事务耗时 logger.info(transaction_format, new Object[] { entry.getHeader().getLogfileName(), - String.valueOf(entry.getHeader().getLogfileOffset()), - String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), - entry.getHeader().getGtid(), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getLogfileOffset()), + String.valueOf(entry.getHeader().getExecuteTime()), + simpleDateFormat.format(date), entry.getHeader().getGtid(), + String.valueOf(delayTime) }); logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId()); printXAInfo(begin.getPropsList()); } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) { @@ -123,9 +117,10 @@ public class BaseCanalClientTest { printXAInfo(end.getPropsList()); logger.info(transaction_format, new Object[] { entry.getHeader().getLogfileName(), - String.valueOf(entry.getHeader().getLogfileOffset()), - String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), - entry.getHeader().getGtid(), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getLogfileOffset()), + String.valueOf(entry.getHeader().getExecuteTime()), + simpleDateFormat.format(date), entry.getHeader().getGtid(), + String.valueOf(delayTime) }); } continue; @@ -143,10 +138,10 @@ public class BaseCanalClientTest { logger.info(row_format, new Object[] { entry.getHeader().getLogfileName(), - String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(), - entry.getHeader().getTableName(), eventType, - String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), - entry.getHeader().getGtid(), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getLogfileOffset()), + entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType, + String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), + entry.getHeader().getGtid(), String.valueOf(delayTime) }); if (eventType == EventType.QUERY || rowChange.getIsDdl()) { logger.info("ddl : " + rowChange.getIsDdl() + " , sql ----> " + rowChange.getSql() + SEP); @@ -174,8 +169,8 @@ public class BaseCanalClientTest { if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB") || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) { // get value bytes - builder.append(column.getName() + " : " - + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8")); + builder.append( + column.getName() + " : " + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8")); } else { builder.append(column.getName() + " : " + column.getValue()); } diff --git a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientPermanceTest.java b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientPermanceTest.java index 2831ad72..311a1147 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientPermanceTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientPermanceTest.java @@ -22,10 +22,8 @@ public class SimpleCanalClientPermanceTest { long end = 0; final ArrayBlockingQueue queue = new ArrayBlockingQueue<>(100); try { - final CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), - destination, - "canal", - "canal"); + final CanalConnector connector = CanalConnectors + .newSingleConnector(new InetSocketAddress(ip, 11111), destination, "canal", "canal"); Thread ackThread = new Thread(() -> { while (true) { diff --git a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java index 21a25e9b..181bfa06 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java @@ -22,10 +22,8 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest { // 根据ip，直接创建链接，无HA的功能 String destination = "example"; String ip = AddressUtils.getHostIp(); - CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), - destination, - "canal", - "canal"); + CanalConnector connector = CanalConnectors + .newSingleConnector(new InetSocketAddress(ip, 11111), destination, "canal", "canal"); final SimpleCanalClientTest clientTest = new SimpleCanalClientTest(destination); clientTest.setConnector(connector); diff --git a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientExample.java b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientExample.java index dfc63179..5ee497a2 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientExample.java @@ -34,7 +34,8 @@ public class CanalKafkaClientExample { public static void main(String[] args) { try { - final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.zkServers, + final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample( + AbstractKafkaTest.zkServers, AbstractKafkaTest.servers, AbstractKafkaTest.topic, AbstractKafkaTest.partition, diff --git a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientFlatMessageExample.java b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientFlatMessageExample.java index 03c5fe8c..0b2c2914 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientFlatMessageExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientFlatMessageExample.java @@ -35,7 +35,8 @@ public class CanalKafkaClientFlatMessageExample { public static void main(String[] args) { try { - final CanalKafkaClientFlatMessageExample kafkaCanalClientExample = new CanalKafkaClientFlatMessageExample(AbstractKafkaTest.zkServers, + final CanalKafkaClientFlatMessageExample kafkaCanalClientExample = new CanalKafkaClientFlatMessageExample( + AbstractKafkaTest.zkServers, AbstractKafkaTest.servers, AbstractKafkaTest.topic, AbstractKafkaTest.partition, diff --git a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaOffsetClientExample.java b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaOffsetClientExample.java index e5732f52..edf41ee8 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaOffsetClientExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaOffsetClientExample.java @@ -43,7 +43,8 @@ public class CanalKafkaOffsetClientExample { public static void main(String[] args) { try { - final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample(AbstractKafkaTest.servers, + final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample( + AbstractKafkaTest.servers, AbstractKafkaTest.topic, AbstractKafkaTest.partition, AbstractKafkaTest.groupId); diff --git a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java index b208d514..cd0a5543 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java @@ -1,13 +1,15 @@ package com.alibaba.otter.canal.example.rocketmq; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.protocol.Message; import java.util.List; import java.util.concurrent.TimeUnit; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.Assert; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; +import com.alibaba.otter.canal.protocol.Message; + /** * RocketMQ client example * @@ -26,15 +28,23 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest { private Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", e); - public CanalRocketMQClientExample(String nameServers, String topic, String groupId) { + public CanalRocketMQClientExample(String nameServers, String topic, String groupId){ connector = new RocketMQCanalConnector(nameServers, topic, groupId, 500, false); } public CanalRocketMQClientExample(String nameServers, String topic, String groupId, boolean enableMessageTrace, - String accessKey, String secretKey, String accessChannel, String namespace) { - connector = new RocketMQCanalConnector(nameServers, topic, groupId, accessKey, - secretKey, -1, false, enableMessageTrace, - null, accessChannel, namespace); + String accessKey, String secretKey, String accessChannel, String namespace){ + connector = new RocketMQCanalConnector(nameServers, + topic, + groupId, + accessKey, + secretKey, + -1, + false, + enableMessageTrace, + null, + accessChannel, + namespace); } public static void main(String[] args) { diff --git a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java index 6f5a73c6..288e032e 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java @@ -18,7 +18,8 @@ import com.alibaba.otter.canal.protocol.FlatMessage; */ public class CanalRocketMQClientFlatMessageExample extends AbstractRocektMQTest { - protected final static Logger logger = LoggerFactory.getLogger(CanalRocketMQClientFlatMessageExample.class); + protected final static Logger logger = LoggerFactory + .getLogger(CanalRocketMQClientFlatMessageExample.class); private RocketMQCanalConnector connector; @@ -34,7 +35,8 @@ public class CanalRocketMQClientFlatMessageExample extends AbstractRocektMQTest public static void main(String[] args) { try { - final CanalRocketMQClientFlatMessageExample rocketMQClientExample = new CanalRocketMQClientFlatMessageExample(nameServers, + final CanalRocketMQClientFlatMessageExample rocketMQClientExample = new CanalRocketMQClientFlatMessageExample( + nameServers, topic, groupId); logger.info("## Start the rocketmq consumer: {}-{}", topic, groupId); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java index dabbd2ed..807ffe49 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java @@ -1,12 +1,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds; -import io.netty.handler.codec.http.HttpResponseStatus; - -import java.io.BufferedOutputStream; -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -import java.io.InputStream; +import java.io.*; import java.util.Comparator; import java.util.Iterator; import java.util.LinkedList; @@ -41,6 +35,8 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile; +import io.netty.handler.codec.http.HttpResponseStatus; + /** * @author chengjin.lyf on 2018/8/7 下午3:10 * @since 1.0.25 @@ -212,7 +208,7 @@ public class BinlogDownloadQueue { InputStream is = response.getEntity().getContent(); boolean isChunked = response.getEntity().isChunked(); Header contentLengthHeader = response.getFirstHeader("Content-Length"); - long totalSize = isChunked || contentLengthHeader == null ? 0 : Long.parseLong(contentLengthHeader.getValue()); + long totalSize = (isChunked || contentLengthHeader == null) ? 0 : Long.parseLong(contentLengthHeader.getValue()); if (response.getFirstHeader("Content-Disposition") != null) { fileName = response.getFirstHeader("Content-Disposition").getValue(); fileName = StringUtils.substringAfter(fileName, "filename=");