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 @@
-
- org.apache.kafka
- kafka-clients
- 0.10.0.1
-
+
+
+
+
+
org.apache.kafka
kafka_2.11
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 record;
if (topic.getPartition() != null) {
record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message);