diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index 7514c5ce..3ef05489 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -91,9 +91,9 @@ public class CanalKafkaStarter { private static void worker(Topic topic) { while (!running) ; while (!CanalServerStarter.isRunning()) ; //等待server启动完成 - logger.info("## start the canal consumer: {}.", topic.getDestination()); + logger.info("## start the canal consumer: {}.", topic.getCanalDestination()); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), (short) 1001, ""); + ClientIdentity clientIdentity = new ClientIdentity(topic.getCanalDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { @@ -105,7 +105,7 @@ public class CanalKafkaStarter { continue; } server.subscribe(clientIdentity); - logger.info("## the canal consumer {} is running now ......", topic.getDestination()); + logger.info("## the canal consumer {} is running now ......", topic.getCanalDestination()); while (running) { Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据 diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java index f70ac2c4..f7362a78 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -21,7 +21,7 @@ public class KafkaProperties { public static class Topic { private String topic; private Integer partition; - private String destination; + private String canalDestination; public String getTopic() { return topic; @@ -39,12 +39,12 @@ public class KafkaProperties { this.partition = partition; } - public String getDestination() { - return destination; + public String getCanalDestination() { + return canalDestination; } - public void setDestination(String destination) { - this.destination = destination; + public void setCanalDestination(String canalDestination) { + this.canalDestination = canalDestination; } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index 3a846c61..7ca1bb7f 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -7,6 +7,6 @@ bufferMemory: 33554432 topics: - topic: example partition: - destination: example + canalDestination: example