diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index 4a93af3d..d2c19448 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -17,13 +17,13 @@ public class CanalClientConfig { private String zookeeperHosts; // 集群模式下的zk地址, 如果配置了单机地址则以单机为准!! - private String mqServers; // kafka or rocket mq 地址 + private String mqServers; // kafka or rocket mq 地址 private Boolean flatMessage = true; // 是否已flatMessage模式传输, 只适用于mq模式 private Integer batchSize; // 批大小 - private Integer retry; // 重试次数 + private Integer retries; // 重试次数 private Long timeout; // 消费超时时间 @@ -79,12 +79,12 @@ public class CanalClientConfig { this.batchSize = batchSize; } - public Integer getRetry() { - return retry; + public Integer getRetries() { + return retries; } - public void setRetry(Integer retry) { - this.retry = retry; + public void setRetries(Integer retries) { + this.retries = retries; } public Long getTimeout() { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 045fee37..8bafcf58 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -45,7 +45,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { while (!running) ; ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetry() == null ? 1 : canalClientConfig.getRetry(); + int retry = canalClientConfig.getRetries() == null ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index b60d66f0..e2d9bea8 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -33,8 +33,6 @@ public class CanalAdapterLoader { private Map canalMQWorker = new HashMap<>(); - private Map outerAdapters = new HashMap<>(); // 配置文件对应adapter - private ExtensionLoader loader; public CanalAdapterLoader(CanalClientConfig canalClientConfig){ diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index 133e4919..5996ab99 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -43,7 +43,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { ; ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetry() == null ? 1 : canalClientConfig.getRetry(); + int retry = canalClientConfig.getRetries() == null ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index 52b4e1af..fabee436 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -64,7 +64,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { ; // waiting until running == true ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetry() == null ? 1 : canalClientConfig.getRetry(); + int retry = canalClientConfig.getRetries() == null ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 300000 : canalClientConfig.getTimeout(); // 默认超时5分钟 Integer batchSize = canalClientConfig.getBatchSize(); if (batchSize == null) { diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 2943d09c..697210f7 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -16,6 +16,8 @@ canal.conf: # zookeeperHosts: slave1:2181 # mqServers: slave1:6667 #or rocketmq # flatMessage: true + retries: + timeout: # srcDataSources: # defaultDS: # url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index f64e6184..80224380 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -85,23 +85,46 @@ public class CanalLauncher { private static MQProperties buildMQPosition(Properties properties) { MQProperties mqProperties = new MQProperties(); - mqProperties.setServers(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS)); - mqProperties.setRetries(Integer.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_RETRIES))); - mqProperties.setBatchSize(Integer.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_BATCHSIZE))); - mqProperties.setLingerMs(Integer.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_LINGERMS))); - mqProperties.setBufferMemory(Long.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_BUFFERMEMORY))); - mqProperties.setCanalBatchSize(Integer.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_CANALBATCHSIZE))); - mqProperties.setCanalGetTimeout(Long.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_CANALGETTIMEOUT))); - mqProperties.setFlatMessage(Boolean.valueOf(CanalController.getProperty(properties, - CanalConstants.CANAL_MQ_FLATMESSAGE))); - mqProperties.setCompressionType(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_COMPRESSION_TYPE)); - mqProperties.setAcks(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_ACKS)); + String servers = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS); + if (!StringUtils.isEmpty(servers)) { + mqProperties.setServers(servers); + } + String retires = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_RETRIES); + if (!StringUtils.isEmpty(retires)) { + mqProperties.setRetries(Integer.valueOf(retires)); + } + String batchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BATCHSIZE); + if (!StringUtils.isEmpty(batchSize)) { + mqProperties.setBatchSize(Integer.valueOf(batchSize)); + } + String lingerMs = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_LINGERMS); + if (!StringUtils.isEmpty(lingerMs)) { + mqProperties.setLingerMs(Integer.valueOf(lingerMs)); + } + String bufferMemory = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BUFFERMEMORY); + if (!StringUtils.isEmpty(bufferMemory)) { + mqProperties.setBufferMemory(Long.valueOf(bufferMemory)); + } + String canalBatchSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALBATCHSIZE); + if (!StringUtils.isEmpty(canalBatchSize)) { + mqProperties.setCanalBatchSize(Integer.valueOf(canalBatchSize)); + } + String canalGetTimeout = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALGETTIMEOUT); + if (!StringUtils.isEmpty(canalGetTimeout)) { + mqProperties.setCanalGetTimeout(Long.valueOf(canalGetTimeout)); + } + String flatMessage = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_FLATMESSAGE); + if (!StringUtils.isEmpty(flatMessage)) { + mqProperties.setFlatMessage(Boolean.valueOf(flatMessage)); + } + String compressionType = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_COMPRESSION_TYPE); + if (!StringUtils.isEmpty(compressionType)) { + mqProperties.setCompressionType(compressionType); + } + String acks = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ACKS); + if (!StringUtils.isEmpty(acks)) { + mqProperties.setAcks(acks); + } return mqProperties; } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 86c700f0..d555c162 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -109,5 +109,5 @@ canal.mq.bufferMemory=33554432 canal.mq.canalBatchSize=50 canal.mq.canalGetTimeout=100 canal.mq.flatMessage=true -#canal.mq.compressionType=none -#canal.mq.acks=all \ No newline at end of file +canal.mq.compressionType= +canal.mq.acks= \ No newline at end of file