From 6faa3fd586eb902bc78d83c0e55ea17b99f15503 Mon Sep 17 00:00:00 2001
From: rewerma <rewerma@163.com>
Date: Tue, 12 Jun 2018 13:28:49 +0800
Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=AF=B9message=E7=9A=84?=
 =?UTF-8?q?=E9=AA=8C=E8=AF=81=EF=BC=8C=E8=BF=87=E6=BB=A4transaction=20begi?=
 =?UTF-8?q?n=20end=20=E7=B1=BB=E5=9E=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 kafka/pom.xml                                       | 10 +++++-----
 .../canal/kafka/producer/CanalKafkaProducer.java    | 13 +++++++++++--
 2 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/kafka/pom.xml b/kafka/pom.xml
index 0ca697ba..55e72553 100644
--- a/kafka/pom.xml
+++ b/kafka/pom.xml
@@ -27,11 +27,11 @@
         </dependency>
 
         <!-- Kafka -->
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-            <version>0.10.0.1</version>
-        </dependency>
+        <!--<dependency>-->
+            <!--<groupId>org.apache.kafka</groupId>-->
+            <!--<artifactId>kafka-clients</artifactId>-->
+            <!--<version>0.10.0.1</version>-->
+        <!--</dependency>-->
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka_2.11</artifactId>
diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
index af2bcfca..21527291 100644
--- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
+++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
@@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.List;
 import java.util.Properties;
 
 /**
@@ -49,10 +48,20 @@ public class CanalKafkaProducer {
     }
 
     public void send(Topic topic, Message message) {
-        if (message == null || message.getEntries().isEmpty()) {
+        boolean valid = false;
+        if (message != null && !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;
         }
 
+
         ProducerRecord<String, Message> record;
         if (topic.getPartition() != null) {
             record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);