chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

去掉kafka server端TRANSACTIONBEGIN和TRANSACTIONEND的判断

Browse Source
This commit is contained in:
machey
2018-09-02 20:21:07 +08:00
parent 21ef2e5dd4
commit ee5775ce05
3 changed files with 5 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -45,7 +45,7 @@ public class KafkaCanalConnector {
properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读 properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置 properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
properties.put("session.timeout.ms", "30000"); // 默认为30秒 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("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", MessageDeserializer.class.getName()); properties.put("value.deserializer", MessageDeserializer.class.getName());

28
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -48,33 +48,7 @@ public class CanalKafkaProducer {
} }
} }
public void send(KafkaProperties.Topic topic, Message message) throws IOException { public void send(KafkaProperties.Topic topic, Message message) {
// 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;
// }
ProducerRecord<String, Message> record; ProducerRecord<String, Message> record;
if (topic.getPartition() != null) { if (topic.getPartition() != null) {
record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message); record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);

6
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java
View File

@@ -54,9 +54,9 @@ public class CanalKafkaStarter implements CanalServerStarter {
canalKafkaProducer = new CanalKafkaProducer(); canalKafkaProducer = new CanalKafkaProducer();
canalKafkaProducer.init(kafkaProperties); canalKafkaProducer.init(kafkaProperties);
// set filterTransactionEntry // set filterTransactionEntry
if (kafkaProperties.isFilterTransactionEntry()) { // if (kafkaProperties.isFilterTransactionEntry()) {
System.setProperty("canal.instance.filter.transaction.entry", "true"); // System.setProperty("canal.instance.filter.transaction.entry", "true");
} // }
// 对应每个instance启动一个worker线程 // 对应每个instance启动一个worker线程
List<CanalDestination> destinations = kafkaProperties.getCanalDestinations(); List<CanalDestination> destinations = kafkaProperties.getCanalDestinations();