diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml
index 2c3e3fff..df9eca81 100644
--- a/kafka/src/main/assembly/dev.xml
+++ b/kafka/src/main/assembly/dev.xml
@@ -34,9 +34,6 @@
**/*
-
- logback.xml
-
./src/main/resources
diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml
index 0b103620..d79ecf3c 100644
--- a/kafka/src/main/assembly/release.xml
+++ b/kafka/src/main/assembly/release.xml
@@ -34,9 +34,6 @@
**/*
-
- logback.xml
-
./src/main/resources
diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
index ae094b27..af2bcfca 100644
--- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
+++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java
@@ -22,9 +22,9 @@ import java.util.Properties;
public class CanalKafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class);
- private static Producer producer;
+ private Producer producer;
- static void init(KafkaProperties kafkaProperties) {
+ public void init(KafkaProperties kafkaProperties) {
Properties properties = new Properties();
properties.put("bootstrap.servers", kafkaProperties.getServers());
properties.put("acks", "all");
@@ -37,7 +37,7 @@ public class CanalKafkaProducer {
producer = new KafkaProducer(properties);
}
- static void stop() {
+ public void stop() {
try {
logger.info("## stop the kafka producer");
producer.close();
@@ -48,29 +48,8 @@ public class CanalKafkaProducer {
}
}
- static void send(Topic topic, Message message) {
- try {
- List entries = message.getEntries();
- boolean flag = false;
- if (!entries.isEmpty()) {
- for (CanalEntry.Entry entry : entries) {
- CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
- if (rowChage.getIsDdl()) {
- flag = true;
- break;
- } else {
- if (!rowChage.getRowDatasList().isEmpty()) {
- flag = true;
- break;
- }
- }
- }
- }
- if (!flag) {
- return;
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
+ public void send(Topic topic, Message message) {
+ if (message == null || message.getEntries().isEmpty()) {
return;
}
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 e89daefc..7514c5ce 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
@@ -1,14 +1,7 @@
package com.alibaba.otter.canal.kafka.producer;
-import java.io.FileInputStream;
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
import com.alibaba.otter.canal.kafka.CanalServerStarter;
import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
-import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
@@ -17,6 +10,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
+import java.io.FileInputStream;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
/**
* kafka 启动类
*
@@ -31,6 +29,8 @@ public class CanalKafkaStarter {
private static ExecutorService executorService;
+ private static CanalKafkaProducer canalKafkaProducer;
+
public static void init() {
try {
@@ -46,7 +46,8 @@ public class CanalKafkaStarter {
}
//初始化 kafka producer
- CanalKafkaProducer.init(kafkaProperties);
+ canalKafkaProducer = new CanalKafkaProducer();
+ canalKafkaProducer.init(kafkaProperties);
//对应每个instance启动一个worker线程
List topics = kafkaProperties.getTopics();
@@ -70,7 +71,7 @@ public class CanalKafkaStarter {
logger.info("## stop the kafka workers");
running = false;
executorService.shutdown();
- CanalKafkaProducer.stop();
+ canalKafkaProducer.stop();
} catch (Throwable e) {
logger.warn("##something goes wrong when stopping kafka workers:", e);
} finally {
@@ -118,7 +119,7 @@ public class CanalKafkaStarter {
//ignore
}
} else {
- CanalKafkaProducer.send(topic, message);
+ canalKafkaProducer.send(topic, message);
}
if (batchId != -1) {
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 4b81238e..f70ac2c4 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
@@ -1,9 +1,7 @@
package com.alibaba.otter.canal.kafka.producer;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
/**
* kafka 配置项
@@ -19,7 +17,6 @@ public class KafkaProperties {
private long bufferMemory = 33554432L;
private List topics = new ArrayList();
- private Map topicMap = new HashMap();
public static class Topic {
private String topic;
@@ -97,15 +94,5 @@ public class KafkaProperties {
public void setTopics(List topics) {
this.topics = topics;
-
- if (topics != null) {
- for (Topic topic : topics) {
- this.topicMap.put(topic.destination, topic);
- }
- }
- }
-
- public Topic getTopicByDestination(String destination) {
- return this.topicMap.get(destination);
}
}
diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml
index af84977d..3a846c61 100644
--- a/kafka/src/main/resources/kafka.yml
+++ b/kafka/src/main/resources/kafka.yml
@@ -5,8 +5,8 @@ lingerMs: 1
bufferMemory: 33554432
topics:
- - topic: example2
+ - topic: example
partition:
- destination: example2
+ destination: example
diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml
deleted file mode 100644
index 158c1699..00000000
--- a/kafka/src/main/resources/logback.xml
+++ /dev/null
@@ -1,85 +0,0 @@
-
-
-
-
- %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
-
-
-
-
-
-
- destination
- canal
-
-
-
- ../logs/${destination}/${destination}.log
-
-
- ../logs/${destination}/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz
-
-
- 512MB
-
- 60
-
-
-
- %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
-
-
-
-
-
-
-
-
- destination
- canal
-
-
-
- ../logs/${destination}/meta.log
-
-
- ../logs/${destination}/%d{yyyy-MM-dd}/meta-%d{yyyy-MM-dd}-%i.log.gz
-
-
- 32MB
-
- 60
-
-
-
- %d{yyyy-MM-dd HH:mm:ss.SSS} - %msg%n
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
\ No newline at end of file