From ee5775ce05acfc053842a69f24ef745c76c74171 Mon Sep 17 00:00:00 2001
From: machey <rewerma@163.com>
Date: Sun, 2 Sep 2018 20:21:07 +0800
Subject: [PATCH] =?UTF-8?q?=E5=8E=BB=E6=8E=89kafka=20server=E7=AB=AFTRANSA?=
 =?UTF-8?q?CTIONBEGIN=E5=92=8CTRANSACTIONEND=E7=9A=84=E5=88=A4=E6=96=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../client/kafka/KafkaCanalConnector.java     |  2 +-
 .../otter/canal/kafka/CanalKafkaProducer.java | 28 +------------------
 .../otter/canal/kafka/CanalKafkaStarter.java  |  6 ++--
 3 files changed, 5 insertions(+), 31 deletions(-)

diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
index c5b2f8fc..7afb9c09 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
+++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
@@ -45,7 +45,7 @@ public class KafkaCanalConnector {
         properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
         properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
         properties.put("session.timeout.ms", "30000"); // 默认为30秒
-        properties.put("max.poll.records", "1"); // 所以一次只取一条数据
+        properties.put("max.poll.records", "1"); // 一次只取一条message数据
         properties.put("key.deserializer", StringDeserializer.class.getName());
         properties.put("value.deserializer", MessageDeserializer.class.getName());
 
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 b6dba051..a6d084e4 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
@@ -48,33 +48,7 @@ public class CanalKafkaProducer {
         }
     }
 
-    public void send(KafkaProperties.Topic topic, Message message) throws IOException {
-        // set canal.instance.filter.transaction.entry = true
-
-        // boolean valid = false;
-        // if (message != null) {
-        // if (message.isRaw() && !message.getRawEntries().isEmpty()) {
-        // for (ByteString byteString : message.getRawEntries()) {
-        // CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString);
-        // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN
-        // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) {
-        // valid = true;
-        // break;
-        // }
-        // }
-        // } else if (!message.getEntries().isEmpty()){
-        // for (CanalEntry.Entry entry : message.getEntries()) {
-        // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN
-        // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) {
-        // valid = true;
-        // break;
-        // }
-        // }
-        // }
-        // }
-        // if (!valid) {
-        // return;
-        // }
+    public void send(KafkaProperties.Topic topic, Message message) {
         ProducerRecord<String, Message> record;
         if (topic.getPartition() != null) {
             record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java
index d244ce31..feef8038 100644
--- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java
+++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java
@@ -54,9 +54,9 @@ public class CanalKafkaStarter implements CanalServerStarter {
             canalKafkaProducer = new CanalKafkaProducer();
             canalKafkaProducer.init(kafkaProperties);
             // set filterTransactionEntry
-            if (kafkaProperties.isFilterTransactionEntry()) {
-                System.setProperty("canal.instance.filter.transaction.entry", "true");
-            }
+            // if (kafkaProperties.isFilterTransactionEntry()) {
+            //     System.setProperty("canal.instance.filter.transaction.entry", "true");
+            // }
             // 对应每个instance启动一个worker线程
             List<CanalDestination> destinations = kafkaProperties.getCanalDestinations();