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配置项属性

Browse Source
This commit is contained in:
rewerma
2018-06-12 10:23:50 +08:00
parent e35a06743c
commit 330da50a5e
3 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -91,9 +91,9 @@ public class CanalKafkaStarter {
private static void worker(Topic topic) { private static void worker(Topic topic) {
while (!running) ; while (!running) ;
while (!CanalServerStarter.isRunning()) ; //等待server启动完成 while (!CanalServerStarter.isRunning()) ; //等待server启动完成
logger.info("## start the canal consumer: {}.", topic.getDestination()); logger.info("## start the canal consumer: {}.", topic.getCanalDestination());
CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); CanalServerWithEmbedded server = CanalServerWithEmbedded.instance();
ClientIdentity clientIdentity = new ClientIdentity(topic.getDestination(), (short) 1001, ""); ClientIdentity clientIdentity = new ClientIdentity(topic.getCanalDestination(), (short) 1001, "");
while (running) { while (running) {
try { try {
if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) {
@@ -105,7 +105,7 @@ public class CanalKafkaStarter {
continue; continue;
} }
server.subscribe(clientIdentity); 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) { while (running) {
Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据 Message message = server.getWithoutAck(clientIdentity, 5 * 1024); // 获取指定数量的数据

10
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java
View File

@@ -21,7 +21,7 @@ public class KafkaProperties {
public static class Topic { public static class Topic {
private String topic; private String topic;
private Integer partition; private Integer partition;
private String destination; private String canalDestination;
public String getTopic() { public String getTopic() {
return topic; return topic;
@@ -39,12 +39,12 @@ public class KafkaProperties {
this.partition = partition; this.partition = partition;
} }
public String getDestination() { public String getCanalDestination() {
return destination; return canalDestination;
} }
public void setDestination(String destination) { public void setCanalDestination(String canalDestination) {
this.destination = destination; this.canalDestination = canalDestination;
} }
} }

2
kafka/src/main/resources/kafka.yml
View File

@@ -7,6 +7,6 @@ bufferMemory: 33554432
topics: topics:
- topic: example - topic: example
partition: partition:
destination: example canalDestination: example