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 record; if (topic.getPartition() != null) { record = new ProducerRecord(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 destinations = kafkaProperties.getCanalDestinations();