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

Merge pull request #1496 from rewerma/master

Browse Source 
远程配置代码整理
This commit is contained in:
agapple
2019-02-18 21:08:02 +08:00
committed by GitHub
parent e07db7b178 05cc627b5d
commit 62338a945e
77 changed files with 7076 additions and 2758 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.common;
import java.util.Properties;
/**
* kafka 配置项
*
@@ -8,22 +10,23 @@ package com.alibaba.otter.canal.common;
*/
public class MQProperties {
private String servers = "127.0.0.1:6667";
private int retries = 0;
private int batchSize = 16384;
private int lingerMs = 1;
private int maxRequestSize = 1048576;
private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true;
private String producerGroup = "Canal-Producer";
private int canalBatchSize = 50;
private Long canalGetTimeout = 100L;
private boolean flatMessage = true;
private String compressionType = "none";
private String acks = "all";
private String aliyunAccessKey = "";
private String aliyunSecretKey = "";
private boolean transaction = false; // 是否开启事务
private String servers = "127.0.0.1:6667";
private int retries = 0;
private int batchSize = 16384;
private int lingerMs = 1;
private int maxRequestSize = 1048576;
private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true;
private String producerGroup = "Canal-Producer";
private int canalBatchSize = 50;
private Long canalGetTimeout = 100L;
private boolean flatMessage = true;
private String compressionType = "none";
private String acks = "all";
private String aliyunAccessKey = "";
private String aliyunSecretKey = "";
private boolean transaction = false; // 是否开启事务
private Properties properties = new Properties();
public static class CanalDestination {
@@ -210,4 +213,12 @@ public class MQProperties {
public void setTransaction(boolean transaction) {
this.transaction = transaction;
}
public Properties getProperties() {
return properties;
}
public void setProperties(Properties properties) {
this.properties = properties;
}
}

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

@@ -49,7 +49,12 @@ public class CanalKafkaProducer implements CanalMQProducer {
properties.put("max.request.size", kafkaProperties.getMaxRequestSize());
properties.put("buffer.memory", kafkaProperties.getBufferMemory());
properties.put("key.serializer", StringSerializer.class.getName());
if(kafkaProperties.getTransaction()){
if (!kafkaProperties.getProperties().isEmpty()) {
properties.putAll(kafkaProperties.getProperties());
}
if (kafkaProperties.getTransaction()) {
properties.put("transactional.id", "canal-transactional-id");
} else {
properties.put("retries", kafkaProperties.getRetries());