fix(canal):fix endless local binlog mode with exception.

This reverts commit 31acfc0101.

README.md

@@ -84,6 +84,7 @@ canal 特别设计了 client-server 模式，交互协议使用 protobuf 3.0 , c
 - canal go客户端: [https://github.com/CanalClient/canal-go](https://github.com/CanalClient/canal-go)
 - canal php客户端: [https://github.com/xingwenge/canal-php](https://github.com/xingwenge/canal-php)
 - canal Python客户端：[https://github.com/haozi3156666/canal-python](https://github.com/haozi3156666/canal-python)
+- canal Rust客户端：[https://github.com/laohanlinux/canal-rs](https://github.com/laohanlinux/canal-rs)
 
 canal 作为 MySQL binlog 增量获取和解析工具，可将变更记录投递到 MQ 系统中，比如 Kafka/RocketMQ，可以借助于 MQ 的多语言能力 
 

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java

@@ -174,11 +174,6 @@ public abstract class AbstractCanalAdapterWorker {
             if (i == retry - 1) {
                 connector.ack();
                 logger.error(e.getMessage() + " Error sync but ACK!");
-                try {
-                    Thread.sleep(500);
-                } catch (InterruptedException e1) {
-                    // ignore
-                }
                 return true;
             } else {
                 connector.rollback();

client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java

@@ -61,11 +61,11 @@ public class KafkaCanalConnector implements CanalMQConnector {
             batchSize = 100;
         }
         properties.put("max.poll.records", batchSize.toString());
-        properties.put("key.deserializer", StringDeserializer.class);
+        properties.put("key.deserializer", StringDeserializer.class.getName());
         if (!flatMessage) {
-            properties.put("value.deserializer", MessageDeserializer.class);
+            properties.put("value.deserializer", MessageDeserializer.class.getName());
         } else {
-            properties.put("value.deserializer", StringDeserializer.class);
+            properties.put("value.deserializer", StringDeserializer.class.getName());
         }
     }
 
@@ -183,13 +183,14 @@ public class KafkaCanalConnector implements CanalMQConnector {
 
         ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
 
+        currentOffsets.clear();
+        for (TopicPartition topicPartition : records.partitions()) {
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
+        }
+
         if (!records.isEmpty()) {
-            currentOffsets.clear();
             List<Message> messages = new ArrayList<>();
             for (ConsumerRecord<String, Message> record : records) {
-                if (currentOffsets.get(record.partition()) == null) {
-                    currentOffsets.put(record.partition(), record.offset());
-                }
                 messages.add(record.value());
             }
             return messages;
@@ -220,13 +221,14 @@ public class KafkaCanalConnector implements CanalMQConnector {
 
         ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
 
+        currentOffsets.clear();
+        for (TopicPartition topicPartition : records.partitions()) {
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
+        }
+
         if (!records.isEmpty()) {
-            currentOffsets.clear();
             List<FlatMessage> flatMessages = new ArrayList<>();
             for (ConsumerRecord<String, String> record : records) {
-                if (currentOffsets.get(record.partition()) == null) {
-                    currentOffsets.put(record.partition(), record.offset());
-                }
                 String flatMessageJson = record.value();
                 FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
                 flatMessages.add(flatMessage);
@@ -246,14 +248,12 @@ public class KafkaCanalConnector implements CanalMQConnector {
         // 回滚所有分区
         if (kafkaConsumer != null) {
             for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
-                kafkaConsumer.commitSync();
+                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
             }
         }
         if (kafkaConsumer2 != null) {
             for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
-                kafkaConsumer.commitSync();
+                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
             }
         }
     }

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

@@ -934,4 +934,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         this.rdsOssMode = rdsOssMode;
     }
 
+    public void setDumpErrorCount(int dumpErrorCount) {
+        this.dumpErrorCount = dumpErrorCount;
+    }
 }

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

@@ -12,11 +12,11 @@ import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
 
 /**
  * aliyun rds的binlog parser支持
- * 
+ *
  * <pre>
  * 注意点：aliyun的binlog会有定期清理并备份到oss上, 这里实现了一份自动下载oss+rds binlog的机制
  * </pre>
- * 
+ *
  * @author chengjin.lyf on 2018/7/20 上午10:52
  * @since 1.0.25
  */
@@ -88,6 +88,9 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
                         @Override
                         public void run() {
                             rdsLocalBinlogEventParser.stop();
+                            // empty the dump error count,or will go into local binlog mode again,with error
+                            // position,never get out,fixed by bucketli
+                            RdsBinlogEventParserProxy.this.setDumpErrorCount(0);
                             RdsBinlogEventParserProxy.this.start();
                         }
                     });