Merge pull request #876 from rewerma/master

整合 kafka 到 server 和 client

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java

@@ -20,9 +20,9 @@ public class CanalClientConfig {
 
     private String              bootstrapServers;
 
-    private List<KafkaTopic>    kafkaTopics    = new ArrayList<>();
+    private List<KafkaTopic>    kafkaTopics;
 
-    private List<CanalInstance> canalInstances = new ArrayList<>();
+    private List<CanalInstance> canalInstances;
 
     public String getCanalServerHost() {
         return canalServerHost;

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java

@@ -45,7 +45,7 @@ public class MappingConfigLoader {
                 continue;
             }
             c = c.trim();
-            if (c.startsWith("#")) {
+            if (c.equals("") || c.startsWith("#")) {
                 continue;
             }
 

client-launcher/pom.xml

@@ -18,17 +18,16 @@
             <artifactId>client-adapter.common</artifactId>
             <version>${project.version}</version>
         </dependency>
-
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.client</artifactId>
             <version>${project.version}</version>
         </dependency>
-
+        <!-- 单独引入kafka依赖 -->
         <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>canal.kafka.client</artifactId>
-            <version>${project.version}</version>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>1.1.1</version>
         </dependency>
         <dependency>
             <groupId>org.yaml</groupId>

client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java

@@ -1,19 +1,25 @@
 package com.alibaba.otter.canal.client.adapter.loader;
 
-import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
-import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker;
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
-import com.alibaba.otter.canal.protocol.Message;
-import org.apache.kafka.clients.consumer.CommitFailedException;
-import org.apache.kafka.common.errors.WakeupException;
-
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.kafka.clients.consumer.CommitFailedException;
+import org.apache.kafka.common.errors.WakeupException;
+
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * kafka对应的client适配器工作线程
+ *
+ * @author machengyuan 2018-8-19 下午11:30:49
+ * @version 1.0.0
+ */
 public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
 
     private KafkaCanalConnector connector;
@@ -85,7 +91,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
     private void process() {
         while (!running)
             ;
-        ExecutorService executor = Executors.newFixedThreadPool(1);
+        ExecutorService executor = Executors.newSingleThreadExecutor();
         final AtomicBoolean executing = new AtomicBoolean(true);
         while (running) {
             try {
@@ -136,7 +142,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
                                 }
                             });
 
-                            while (executing.get()) { // keeping kafka client active
+                            // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
+                            while (executing.get()) {
                                 connector.ack();
                                 Thread.sleep(500);
                             }

client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java

@@ -26,13 +26,13 @@ public class CanalAdapterLoader {
 
     private static final Logger                  logger            = LoggerFactory.getLogger(CanalAdapterLoader.class);
 
-    private CanalClientConfig canalClientConfig;
+    private CanalClientConfig                    canalClientConfig;
 
     private Map<String, CanalAdapterWorker>      canalWorkers      = new HashMap<>();
 
     private Map<String, CanalAdapterKafkaWorker> canalKafkaWorkers = new HashMap<>();
 
-    private ExtensionLoader<CanalOuterAdapter> loader;
+    private ExtensionLoader<CanalOuterAdapter>   loader;
 
     public CanalAdapterLoader(CanalClientConfig canalClientConfig){
         this.canalClientConfig = canalClientConfig;
@@ -43,7 +43,7 @@ public class CanalAdapterLoader {
      */
     public void init() {
         // canal instances 和 kafka topics 配置不能同时为空
-        if (canalClientConfig.getCanalInstances().isEmpty() && canalClientConfig.getKafkaTopics().isEmpty()) {
+        if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getKafkaTopics() == null) {
             throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics");
         }
 
@@ -58,56 +58,61 @@ public class CanalAdapterLoader {
         }
         String zkHosts = this.canalClientConfig.getZookeeperHosts();
 
-        if (zkHosts == null && sa == null) {
-            throw new RuntimeException("Blank config property: canalServerHost or zookeeperHosts");
-        }
+        // if (zkHosts == null && sa == null) {
+        // throw new RuntimeException("Blank config property: canalServerHost or
+        // zookeeperHosts");
+        // }
 
         // 初始化canal-client的适配器
-        for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) {
-            List<List<CanalOuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
+        if (canalClientConfig.getCanalInstances() != null) {
+            for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) {
+                List<List<CanalOuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
 
-            for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) {
-                List<CanalOuterAdapter> canalOutConnectors = new ArrayList<>();
-                for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) {
-                    loadConnector(c, canalOutConnectors);
+                for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) {
+                    List<CanalOuterAdapter> canalOutConnectors = new ArrayList<>();
+                    for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) {
+                        loadConnector(c, canalOutConnectors);
+                    }
+                    canalOuterAdapterGroups.add(canalOutConnectors);
                 }
-                canalOuterAdapterGroups.add(canalOutConnectors);
-            }
-            CanalAdapterWorker worker;
-            if (zkHosts != null) {
-                worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups);
-            } else {
-                worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups);
+                CanalAdapterWorker worker;
+                if (zkHosts != null) {
+                    worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups);
+                } else {
+                    worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups);
+                }
+                canalWorkers.put(instance.getInstance(), worker);
+                worker.start();
+                logger.info("Start adapter for canal instance: {} succeed", instance.getInstance());
             }
-            canalWorkers.put(instance.getInstance(), worker);
-            worker.start();
-            logger.info("Start adapter for canal instance: {} succeed", instance.getInstance());
         }
 
         // 初始化canal-client-kafka的适配器
-        for (CanalClientConfig.KafkaTopic kafkaTopic : canalClientConfig.getKafkaTopics()) {
-            for (CanalClientConfig.Group group : kafkaTopic.getGroups()) {
-                List<List<CanalOuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
+        if (canalClientConfig.getKafkaTopics() != null) {
+            for (CanalClientConfig.KafkaTopic kafkaTopic : canalClientConfig.getKafkaTopics()) {
+                for (CanalClientConfig.Group group : kafkaTopic.getGroups()) {
+                    List<List<CanalOuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
 
-                List<CanalOuterAdapter> canalOuterAdapters = new ArrayList<>();
+                    List<CanalOuterAdapter> canalOuterAdapters = new ArrayList<>();
 
-                for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) {
-                    // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) {
-                    loadConnector(config, canalOuterAdapters);
-                    // }
+                    for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) {
+                        // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) {
+                        loadConnector(config, canalOuterAdapters);
+                        // }
+                    }
+                    canalOuterAdapterGroups.add(canalOuterAdapters);
+
+                    // String zkServers = canalClientConfig.getZookeeperHosts();
+                    CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkHosts,
+                        canalClientConfig.getBootstrapServers(),
+                        kafkaTopic.getTopic(),
+                        group.getGroupId(),
+                        canalOuterAdapterGroups);
+                    canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker);
+                    canalKafkaWorker.start();
+                    logger.info("Start adapter for canal-client kafka topic: {} succeed",
+                        kafkaTopic.getTopic() + "-" + group.getGroupId());
                 }
-                canalOuterAdapterGroups.add(canalOuterAdapters);
-
-                String zkServers = canalClientConfig.getZookeeperHosts();
-                CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkServers,
-                    canalClientConfig.getBootstrapServers(),
-                    kafkaTopic.getTopic(),
-                    group.getGroupId(),
-                    canalOuterAdapterGroups);
-                canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker);
-                canalKafkaWorker.start();
-                logger.info("Start adapter for canal-client kafka topic: {} succeed",
-                    kafkaTopic.getTopic() + "-" + group.getGroupId());
             }
         }
     }

client-launcher/src/main/resources/canal-client.yml

@@ -1,6 +1,6 @@
 canalServerHost: 127.0.0.1:11111
-#zookeeperHosts: 127.0.0.1:2181
-#bootstrapServers: kafka1.mytest.com:9092,kafka2.mytest.com:9092
+#zookeeperHosts: slave1:2181
+#bootstrapServers: slave1:6667,slave2:6667
 
 canalInstances:
 - instance: example
@@ -10,13 +10,12 @@ canalInstances:
     - name: hbase
       hosts: slave1:2181
       properties: {znodeParent: "/hbase-unsecure"}
-
 #kafkaTopics:
-#- topic: devmysql4308
+#- topic: example
 #  groups:
-#  - groupId: devmysql4308_es
-#    adapters:
-#    - name: es
-#      hosts:
-#      zkHosts:
-#      properties: {clusterName: es-service-test}
+#  - groupId: example_g1
+#    outAdapters:
+#    - name: logger
+#    - name: hbase
+#      hosts: slave1:2181
+#      properties: {znodeParent: "/hbase-unsecure"}

client/pom.xml

@@ -16,6 +16,13 @@
 			<artifactId>canal.protocol</artifactId>
 			<version>${project.version}</version>
 		</dependency>
+		<!-- 客户端要使用请单独引入kafka依赖 -->
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-clients</artifactId>
+			<version>1.1.1</version>
+			<scope>provided</scope>
+		</dependency>
 
 		<!-- junit -->
 		<dependency>

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java

@@ -1,21 +1,19 @@
-package com.alibaba.otter.canal.kafka.client;
+package com.alibaba.otter.canal.client.kafka;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
-import com.alibaba.otter.canal.common.utils.AddressUtils;
-import com.alibaba.otter.canal.common.utils.BooleanMutex;
-import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
-import com.alibaba.otter.canal.kafka.client.running.ClientRunningData;
-import com.alibaba.otter.canal.kafka.client.running.ClientRunningListener;
-import com.alibaba.otter.canal.kafka.client.running.ClientRunningMonitor;
-import com.alibaba.otter.canal.protocol.Message;
-import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
-import java.util.Collections;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
+import com.alibaba.otter.canal.client.kafka.running.ClientRunningData;
+import com.alibaba.otter.canal.common.utils.AddressUtils;
+import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
+import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 
 /**
  * canal kafka 数据操作客户端
@@ -26,16 +24,16 @@ import java.util.concurrent.TimeUnit;
 public class KafkaCanalConnector {
 
     private KafkaConsumer<String, Message> kafkaConsumer;
-    private String topic;
-    private Integer partition;
-    private Properties properties;
-    private ClientRunningMonitor runningMonitor;  // 运行控制
-    private ZkClientx zkClientx;
-    private BooleanMutex mutex = new BooleanMutex(false);
-    private volatile boolean connected = false;
-    private volatile boolean running = false;
-
-    public KafkaCanalConnector(String zkServers, String servers, String topic, Integer partition, String groupId) {
+    private String                         topic;
+    private Integer                        partition;
+    private Properties                     properties;
+    // private ClientRunningMonitor runningMonitor; // 运行控制
+    // private BooleanMutex mutex = new BooleanMutex(false);
+    private ZkClientx                      zkClientx;
+    private volatile boolean               connected = false;
+    private volatile boolean               running   = false;
+
+    public KafkaCanalConnector(String zkServers, String servers, String topic, Integer partition, String groupId){
         this.topic = topic;
         this.partition = partition;
 
@@ -44,7 +42,7 @@ public class KafkaCanalConnector {
         properties.put("group.id", groupId);
         properties.put("enable.auto.commit", false);
         properties.put("auto.commit.interval.ms", "1000");
-        properties.put("auto.offset.reset", "latest"); //如果没有offset则从最后的offset开始读
+        properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读
         properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
         properties.put("session.timeout.ms", "30000"); // 默认为30秒
         properties.put("max.poll.records", "1"); // 所以一次只取一条数据
@@ -58,19 +56,19 @@ public class KafkaCanalConnector {
             clientData.setGroupId(groupId);
             clientData.setAddress(AddressUtils.getHostIp());
 
-            runningMonitor = new ClientRunningMonitor();
-            runningMonitor.setTopic(topic);
-            runningMonitor.setZkClient(zkClientx);
-            runningMonitor.setClientData(clientData);
-            runningMonitor.setListener(new ClientRunningListener() {
-                public void processActiveEnter() {
-                    mutex.set(true);
-                }
-
-                public void processActiveExit() {
-                    mutex.set(false);
-                }
-            });
+            // runningMonitor = new ClientRunningMonitor();
+            // runningMonitor.setTopic(topic);
+            // runningMonitor.setZkClient(zkClientx);
+            // runningMonitor.setClientData(clientData);
+            // runningMonitor.setListener(new ClientRunningListener() {
+            // public void processActiveEnter() {
+            // mutex.set(true);
+            // }
+            //
+            // public void processActiveExit() {
+            // mutex.set(false);
+            // }
+            // });
         }
 
     }
@@ -95,11 +93,11 @@ public class KafkaCanalConnector {
             return;
         }
 
-        if (runningMonitor != null) {
-            if (!runningMonitor.isStart()) {
-                runningMonitor.start();
-            }
-        }
+        // if (runningMonitor != null) {
+        // if (!runningMonitor.isStart()) {
+        // runningMonitor.start();
+        // }
+        // }
 
         connected = true;
 
@@ -115,9 +113,9 @@ public class KafkaCanalConnector {
         kafkaConsumer.close();
 
         connected = false;
-        if (runningMonitor.isStart()) {
-            runningMonitor.stop();
-        }
+        // if (runningMonitor.isStart()) {
+        // runningMonitor.stop();
+        // }
     }
 
     private void waitClientRunning() {
@@ -128,12 +126,12 @@ public class KafkaCanalConnector {
                 }
 
                 running = true;
-                mutex.get();// 阻塞等待
+                // mutex.get();// 阻塞等待
             } else {
                 // 单机模式直接设置为running
                 running = true;
             }
-        } catch (InterruptedException e) {
+        } catch (Exception e) {
             Thread.currentThread().interrupt();
             throw new CanalClientException(e);
         }
@@ -141,7 +139,8 @@ public class KafkaCanalConnector {
 
     public boolean checkValid() {
         if (zkClientx != null) {
-            return mutex.state();
+            // return mutex.state();
+            return true;
         } else {
             return true;// 默认都放过
         }
@@ -234,9 +233,9 @@ public class KafkaCanalConnector {
     public void stopRunning() {
         if (running) {
             running = false; // 设置为非running状态
-            if (!mutex.state()) {
-                mutex.set(true); // 中断阻塞
-            }
+            // if (!mutex.state()) {
+            // mutex.set(true); // 中断阻塞
+            // }
         }
     }
 }

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client;
+package com.alibaba.otter.canal.client.kafka;
 
 /**
  * canal kafka connectors创建工具类

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client;
+package com.alibaba.otter.canal.client.kafka;
 
 import java.util.Map;
 

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningData.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/running/ClientRunningData.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.kafka.running;
 
 /**
  * client running状态信息

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningListener.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/running/ClientRunningListener.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.kafka.running;
 
 /**
  * client running状态信息

kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningMonitor.java → client/src/main/java/com/alibaba/otter/canal/client/kafka/running/ClientRunningMonitor.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.kafka.running;
 
 import java.text.MessageFormat;
 import java.util.Random;
@@ -55,15 +55,15 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
     }
 
     private static final Logger        logger       = LoggerFactory.getLogger(ClientRunningMonitor.class);
-    private ZkClientx                  zkClient;
+    private ZkClientx zkClient;
     private String                     topic;
-    private ClientRunningData          clientData;
+    private ClientRunningData clientData;
     private IZkDataListener            dataListener;
-    private BooleanMutex               mutex        = new BooleanMutex(false);
+    private BooleanMutex mutex        = new BooleanMutex(false);
     private volatile boolean           release      = false;
     private volatile ClientRunningData activeData;
     private ScheduledExecutorService   delayExector = Executors.newScheduledThreadPool(1);
-    private ClientRunningListener      listener;
+    private ClientRunningListener listener;
     private int                        delayTime    = 5;
 
     private static Integer             virtualPort;

kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java → client/src/test/java/com/alibaba/otter/canal/client/running/kafka/AbstractKafkaTest.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.running.kafka;
 
 import org.junit.Assert;
 

kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java → client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.running.kafka;
 
 import java.util.concurrent.TimeUnit;
 
@@ -7,8 +7,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
 
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**

kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java → client/src/test/java/com/alibaba/otter/canal/client/running/kafka/KafkaClientRunningTest.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.client.running;
+package com.alibaba.otter.canal.client.running.kafka;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -9,8 +9,8 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
-import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**

deployer/pom.xml

@@ -40,6 +40,7 @@
 						<exclude>**/canal.properties</exclude>
 						<exclude>**/spring/**</exclude>
 						<exclude>**/example/**</exclude>
+						<exclude>**/kafka.yml</exclude>
 					</excludes>
 				</configuration>
 			</plugin>

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java

@@ -23,6 +23,7 @@ public class CanalConstants {
     public static final String CANAL_AUTO_SCAN                   = ROOT + "." + "auto.scan";
     public static final String CANAL_AUTO_SCAN_INTERVAL          = ROOT + "." + "auto.scan.interval";
     public static final String CANAL_CONF_DIR                    = ROOT + "." + "conf.dir";
+    public static final String CANAL_SERVER_MODE                 = ROOT + "." + "serverMode";
 
     public static final String CANAL_DESTINATION_SPLIT           = ",";
     public static final String GLOBAL_NAME                       = "global";

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java

@@ -3,6 +3,8 @@ package com.alibaba.otter.canal.deployer;
 import java.io.FileInputStream;
 import java.util.Properties;
 
+import com.alibaba.otter.canal.kafka.CanalKafkaStarter;
+import com.alibaba.otter.canal.server.CanalServerStarter;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,6 +53,17 @@ public class CanalLauncher {
                 }
 
             });
+
+            CanalServerStarter canalServerStarter = null;
+            String serverMode = properties.getProperty(CanalConstants.CANAL_SERVER_MODE, "tcp");
+            if (serverMode.equalsIgnoreCase("kafka")) {
+                canalServerStarter = new CanalKafkaStarter();
+            } else if (serverMode.equalsIgnoreCase("rocketMQ")) {
+                // 预留rocketMQ启动
+            }
+            if (canalServerStarter != null) {
+                canalServerStarter.init();
+            }
         } catch (Throwable e) {
             logger.error("## Something goes wrong when starting up the canal Server:", e);
             System.exit(0);

deployer/src/main/resources/canal.properties

@@ -9,6 +9,8 @@ canal.zkServers=
 # flush data to zk
 canal.zookeeper.flush.period = 1000
 canal.withoutNetty = false
+# tcp, kafka, rocketMQ
+canal.serverMode = tcp
 # flush meta cursor/parse position to file
 canal.file.data.dir = ${canal.conf.dir}
 canal.file.flush.period = 1000

kafka/src/main/resources/kafka.yml → deployer/src/main/resources/kafka.yml

@@ -12,8 +12,8 @@ canalDestinations:
     topic: example
     partition:
     # 一个destination可以对应多个topic
-#    topics:
-#      - topic: example
-#        partition:
+    #topics:
+    #  - topic: example
+    #    partition:
 
 

kafka-client/pom.xml

@@ -1,109 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <artifactId>canal</artifactId>
-        <groupId>com.alibaba.otter</groupId>
-        <version>1.1.1-SNAPSHOT</version>
-        <relativePath>../pom.xml</relativePath>
-    </parent>
-    <groupId>com.alibaba.otter</groupId>
-    <artifactId>canal.kafka.client</artifactId>
-    <packaging>jar</packaging>
-    <name>canal kafka client module for otter ${project.version}</name>
-
-
-    <dependencies>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>canal.protocol</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-            <version>1.1.1</version>
-        </dependency>
-
-        <!-- junit -->
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-        </dependency>
-    </dependencies>
-
-    <profiles>
-        <profile>
-            <id>dev</id>
-            <activation>
-                <activeByDefault>true</activeByDefault>
-                <property>
-                    <name>env</name>
-                    <value>!javadoc</value>
-                </property>
-            </activation>
-        </profile>
-
-        <profile>
-            <id>javadoc</id>
-            <activation>
-                <property>
-                    <name>env</name>
-                    <value>javadoc</value>
-                </property>
-            </activation>
-            <build>
-                <plugins>
-                    <plugin>
-                        <groupId>org.apache.maven.plugins</groupId>
-                        <artifactId>maven-javadoc-plugin</artifactId>
-                        <version>2.9.1</version>
-                        <executions>
-                            <execution>
-                                <id>attach-javadocs</id>
-                                <phase>package</phase>
-                                <goals>
-                                    <goal>jar</goal>
-                                </goals>
-                            </execution>
-                        </executions>
-                        <configuration>
-                            <aggregate>true</aggregate>
-                            <show>public</show>
-                            <nohelp>true</nohelp>
-                            <header>${project.artifactId}-${project.version}</header>
-                            <footer>${project.artifactId}-${project.version}</footer>
-                            <doctitle>${project.artifactId}-${project.version}</doctitle>
-                            <links>
-                                <link>https://github.com/alibaba/canal</link>
-                            </links>
-                            <outputDirectory>${project.build.directory}/apidocs/apidocs/${project.version}</outputDirectory>
-                        </configuration>
-                    </plugin>
-                    <plugin>
-                        <groupId>org.apache.maven.plugins</groupId>
-                        <artifactId>maven-scm-publish-plugin</artifactId>
-                        <version>1.0-beta-2</version>
-                        <executions>
-                            <execution>
-                                <id>attach-javadocs</id>
-                                <phase>package</phase>
-                                <goals>
-                                    <goal>publish-scm</goal>
-                                </goals>
-                            </execution>
-                        </executions>
-                        <configuration>
-                            <checkoutDirectory>${project.build.directory}/scmpublish</checkoutDirectory>
-                            <checkinComment>Publishing javadoc for ${project.artifactId}:${project.version}</checkinComment>
-                            <content>${project.build.directory}/apidocs</content>
-                            <skipDeletedFiles>true</skipDeletedFiles>
-                            <pubScmUrl>scm:git:git@github.com:alibaba/canal.git</pubScmUrl>
-                            <scmBranch>gh-pages</scmBranch>
-                        </configuration>
-                    </plugin>
-                </plugins>
-            </build>
-        </profile>
-    </profiles>
-</project>

kafka-client/src/test/resources/logback.xml

@@ -1,19 +0,0 @@
-<configuration scan="true" scanPeriod=" 5 seconds">
-
-	<jmxConfigurator />
-	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
-		<encoder>
-			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
-			</pattern>
-		</encoder>
-	</appender>
-
-	<logger name="org.apache.kafka" additivity="false">
-		<level value="ERROR" />
-		<appender-ref ref="STDOUT" />
-	</logger>
-
-	<root level="INFO">
-		<appender-ref ref="STDOUT"/>
-	</root>
-</configuration>

kafka/pom.xml

@@ -1,143 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <artifactId>canal</artifactId>
-        <groupId>com.alibaba.otter</groupId>
-        <version>1.1.1-SNAPSHOT</version>
-        <relativePath>../pom.xml</relativePath>
-    </parent>
-    <groupId>com.alibaba.otter</groupId>
-    <artifactId>canal.kafka</artifactId>
-    <packaging>jar</packaging>
-    <name>canal kafka module for otter ${project.version}</name>
-    <dependencies>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>canal.deployer</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.yaml</groupId>
-            <artifactId>snakeyaml</artifactId>
-            <version>1.17</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka_2.11</artifactId>
-            <version>1.1.1</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.slf4j</groupId>
-                    <artifactId>slf4j-log4j12</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-        <!--kafka_2.11_1.1.1 exclusion掉了netty 的依赖，但CanalServerWithNetty 依赖 netty3,升级kafka至 1.1.1 需要显示加入,否则会启动失败 -->
-        <dependency>
-            <groupId>org.jboss.netty</groupId>
-            <artifactId>netty</artifactId>
-            <version>3.2.2.Final</version>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <!-- deploy模块的packaging通常是jar，如果项目中没有java 源代码或资源文件，加上这一段配置使项目能通过构建 -->
-            <plugin>
-                <artifactId>maven-jar-plugin</artifactId>
-                <configuration>
-                    <archive>
-                        <addMavenDescriptor>true</addMavenDescriptor>
-                    </archive>
-                    <excludes>
-                        <exclude>**/logback.xml</exclude>
-                        <exclude>**/canal.properties</exclude>
-                        <exclude>**/spring/**</exclude>
-                        <exclude>**/example/**</exclude>
-                        <exclude>**/kafka.yml</exclude>
-                    </excludes>
-                </configuration>
-            </plugin>
-
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <!-- 这是最新版本，推荐使用这个版本 -->
-                <version>2.2.1</version>
-                <executions>
-                    <execution>
-                        <id>assemble</id>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                        <phase>package</phase>
-                    </execution>
-                </executions>
-                <configuration>
-                    <appendAssemblyId>false</appendAssemblyId>
-                    <attach>false</attach>
-                </configuration>
-            </plugin>
-        </plugins>
-    </build>
-
-    <profiles>
-        <profile>
-            <id>dev</id>
-            <activation>
-                <activeByDefault>true</activeByDefault>
-                <property>
-                    <name>env</name>
-                    <value>!release</value>
-                </property>
-            </activation>
-
-            <build>
-                <plugins>
-                    <plugin>
-                        <artifactId>maven-assembly-plugin</artifactId>
-                        <configuration>
-                            <!-- maven assembly插件需要一个描述文件 来告诉插件包的结构以及打包所需的文件来自哪里 -->
-                            <descriptors>
-                                <descriptor>${basedir}/src/main/assembly/dev.xml</descriptor>
-                            </descriptors>
-                            <finalName>canal</finalName>
-                            <outputDirectory>${project.build.directory}</outputDirectory>
-                        </configuration>
-                    </plugin>
-                </plugins>
-            </build>
-
-        </profile>
-
-        <profile>
-            <id>release</id>
-            <activation>
-                <property>
-                    <name>env</name>
-                    <value>release</value>
-                </property>
-            </activation>
-
-            <build>
-                <plugins>
-                    <plugin>
-                        <artifactId>maven-assembly-plugin</artifactId>
-                        <configuration>
-                            <!-- 发布模式使用的maven assembly插件描述文件 -->
-                            <descriptors>
-                                <descriptor>${basedir}/src/main/assembly/release.xml</descriptor>
-                            </descriptors>
-                            <!-- 如果一个应用的包含多个deploy模块，如果使用同样的包名， 如果把它们复制的一个目录中可能会失败，所以包名加了 artifactId以示区分 -->
-                            <finalName>${project.artifactId}-${project.version}</finalName>
-                            <!-- scm 要求 release 模式打出的包放到顶级目录下的target子目录中 -->
-                            <outputDirectory>${project.parent.build.directory}</outputDirectory>
-                        </configuration>
-                    </plugin>
-                </plugins>
-            </build>
-        </profile>
-    </profiles>
-</project>

kafka/src/main/assembly/dev.xml

@@ -1,64 +0,0 @@
-<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
-	<id>dist</id>
-	<formats>
-		<format>dir</format>
-	</formats>
-	<includeBaseDirectory>false</includeBaseDirectory>
-	<fileSets>
-		<fileSet>
-			<directory>.</directory>
-			<outputDirectory>/</outputDirectory>
-			<includes>
-				<include>README*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/bin</directory>
-			<outputDirectory>bin</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<fileMode>0755</fileMode>
-		</fileSet>
-		<fileSet>
-			<directory>../deployer/src/main/conf</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>../deployer/src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<excludes>
-				<exclude>logback.xml</exclude>
-			</excludes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>target</directory>
-			<outputDirectory>logs</outputDirectory>
-			<excludes>
-				<exclude>**/*</exclude>
-			</excludes>
-		</fileSet>
-	</fileSets>
-	<dependencySets>
-		<dependencySet>
-			<outputDirectory>lib</outputDirectory>
-			<excludes>
-				<exclude>junit:junit</exclude>
-			</excludes>
-		</dependencySet>
-	</dependencySets>
-</assembly>

kafka/src/main/assembly/release.xml

@@ -1,64 +0,0 @@
-<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
-	<id>dist</id>
-	<formats>
-		<format>tar.gz</format>
-	</formats>
-	<includeBaseDirectory>false</includeBaseDirectory>
-	<fileSets>
-		<fileSet>
-			<directory>.</directory>
-			<outputDirectory>/</outputDirectory>
-			<includes>
-				<include>README*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/bin</directory>
-			<outputDirectory>bin</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<fileMode>0755</fileMode>
-		</fileSet>
-		<fileSet>
-			<directory>../deployer/src/main/conf</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>../deployer/src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<excludes>
-				<exclude>logback.xml</exclude>
-			</excludes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>target</directory>
-			<outputDirectory>logs</outputDirectory>
-			<excludes>
-				<exclude>**/*</exclude>
-			</excludes>
-		</fileSet>
-	</fileSets>
-	<dependencySets>
-		<dependencySet>
-			<outputDirectory>lib</outputDirectory>
-			<excludes>
-				<exclude>junit:junit</exclude>
-			</excludes>
-		</dependencySet>
-	</dependencySets>
-</assembly>

kafka/src/main/bin/startup.bat

@@ -1,25 +0,0 @@
-@echo off
-@if not "%ECHO%" == ""  echo %ECHO%
-@if "%OS%" == "Windows_NT"  setlocal
-
-set ENV_PATH=.\
-if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0%
-
-set conf_dir=%ENV_PATH%\..\conf
-set canal_conf=%conf_dir%\canal.properties
-set logback_configurationFile=%conf_dir%\logback.xml
-
-set CLASSPATH=%conf_dir%
-set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
-
-set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m
-set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
-set JAVA_DEBUG_OPT= -server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n
-set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%"
-
-set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %JAVA_DEBUG_OPT% %CANAL_OPTS%
-
-set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.kafka.CanalLauncher
-echo start cmd : %CMD_STR%
-
-java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.kafka.CanalLauncher

kafka/src/main/bin/startup.sh

@@ -1,104 +0,0 @@
-#!/bin/bash 
-
-current_path=`pwd`
-case "`uname`" in
-    Linux)
-		bin_abs_path=$(readlink -f $(dirname $0))
-		;;
-	*)
-		bin_abs_path=`cd $(dirname $0); pwd`
-		;;
-esac
-base=${bin_abs_path}/..
-canal_conf=$base/conf/canal.properties
-logback_configurationFile=$base/conf/logback.xml
-export LANG=en_US.UTF-8
-export BASE=$base
-
-if [ -f $base/bin/canal.pid ] ; then
-	echo "found canal.pid , Please run stop.sh first ,then startup.sh" 2>&2
-    exit 1
-fi
-
-if [ ! -d $base/logs/canal ] ; then 
-	mkdir -p $base/logs/canal
-fi
-
-## set java path
-if [ -z "$JAVA" ] ; then
-  JAVA=$(which java)
-fi
-
-ALIBABA_JAVA="/usr/alibaba/java/bin/java"
-TAOBAO_JAVA="/opt/taobao/java/bin/java"
-if [ -z "$JAVA" ]; then
-  if [ -f $ALIBABA_JAVA ] ; then
-  	JAVA=$ALIBABA_JAVA
-  elif [ -f $TAOBAO_JAVA ] ; then
-  	JAVA=$TAOBAO_JAVA
-  else
-  	echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2
-    exit 1
-  fi
-fi
-
-case "$#" 
-in
-0 ) 
-	;;
-1 )	
-	var=$*
-	if [ -f $var ] ; then 
-		canal_conf=$var
-	else
-		echo "THE PARAMETER IS NOT CORRECT.PLEASE CHECK AGAIN."
-        exit
-	fi;;
-2 )	
-	var=$1
-	if [ -f $var ] ; then
-		canal_conf=$var
-	else 
-		if [ "$1" = "debug" ]; then
-			DEBUG_PORT=$2
-			DEBUG_SUSPEND="n"
-			JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND"
-		fi
-     fi;;
-* )
-	echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN."
-	exit;;
-esac
-
-str=`file -L $JAVA | grep 64-bit`
-if [ -n "$str" ]; then
-	JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError"
-else
-	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m "
-fi
-
-JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8"
-CANAL_OPTS="-DappName=otter-canal -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf"
-
-if [ -e $canal_conf -a -e $logback_configurationFile ]
-then 
-	
-	for i in $base/lib/*;
-		do CLASSPATH=$i:"$CLASSPATH";
-	done
- 	CLASSPATH="$base/conf:$CLASSPATH";
- 	
- 	echo "cd to $bin_abs_path for workaround relative path"
-  	cd $bin_abs_path
- 	
-	echo LOG CONFIGURATION : $logback_configurationFile
-	echo canal conf : $canal_conf 
-	echo CLASSPATH :$CLASSPATH
-	$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.kafka.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 &
-	echo $! > $base/bin/canal.pid 
-	
-	echo "cd to $current_path for continue"
-  	cd $current_path
-else 
-	echo "canal conf("$canal_conf") OR log configration file($logback_configurationFile) is not exist,please create then first!"
-fi

kafka/src/main/bin/stop.sh

@@ -1,65 +0,0 @@
-#!/bin/bash
-
-cygwin=false;
-linux=false;
-case "`uname`" in
-    CYGWIN*)
-        cygwin=true
-        ;;
-    Linux*)
-    	linux=true
-    	;;
-esac
-
-get_pid() {	
-	STR=$1
-	PID=$2
-    if $cygwin; then
-        JAVA_CMD="$JAVA_HOME\bin\java"
-        JAVA_CMD=`cygpath --path --unix $JAVA_CMD`
-        JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'`
-    else
-    	if $linux; then
-	        if [ ! -z "$PID" ]; then
-	        	JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'`
-		    else 
-		        JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'`
-	        fi
-	    else
-	    	if [ ! -z "$PID" ]; then
-	        	JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'`
-		    else 
-		        JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'`
-	        fi
-	    fi
-    fi
-    echo $JAVA_PID;
-}
-
-base=`dirname $0`/..
-pidfile=$base/bin/canal.pid
-if [ ! -f "$pidfile" ];then
-	echo "canal is not running. exists"
-	exit
-fi
-
-pid=`cat $pidfile`
-if [ "$pid" == "" ] ; then
-	pid=`get_pid "appName=otter-canal"`
-fi
-
-echo -e "`hostname`: stopping canal $pid ... "
-kill $pid
-
-LOOPS=0
-while (true); 
-do 
-	gpid=`get_pid "appName=otter-canal" "$pid"`
-    if [ "$gpid" == "" ] ; then
-    	echo "Oook! cost:$LOOPS"
-    	`rm $pidfile`
-    	break;
-    fi
-    let LOOPS=LOOPS+1
-    sleep 1
-done

kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalLauncher.java

@@ -1,17 +0,0 @@
-package com.alibaba.otter.canal.kafka;
-
-import com.alibaba.otter.canal.kafka.producer.CanalKafkaStarter;
-
-/**
- * canal-kafka独立版本启动的入口类
- *
- * @author machengyuan 2018-6-11 下午05:30:49
- * @version 1.0.0
- */
-public class CanalLauncher {
-
-    public static void main(String[] args) {
-        CanalServerStarter.init();
-        CanalKafkaStarter.init();
-    }
-}

kafka/src/main/java/com/alibaba/otter/canal/kafka/CanalServerStarter.java

@@ -1,78 +0,0 @@
-package com.alibaba.otter.canal.kafka;
-
-import java.io.FileInputStream;
-import java.util.Properties;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.deployer.CanalController;
-
-/**
- * canal server 启动类
- *
- * @author machengyuan 2018-6-11 下午05:30:49
- * @version 1.0.0
- */
-public class CanalServerStarter {
-
-    private static final String     CLASSPATH_URL_PREFIX = "classpath:";
-    private static final Logger     logger               = LoggerFactory.getLogger(CanalServerStarter.class);
-    private volatile static boolean running              = false;
-
-    public static void init() {
-        try {
-            logger.info("## set default uncaught exception handler");
-            setGlobalUncaughtExceptionHandler();
-
-            logger.info("## load canal configurations");
-            String conf = System.getProperty("canal.conf", "classpath:canal.properties");
-            Properties properties = new Properties();
-            if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
-                conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
-                properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf));
-            } else {
-                properties.load(new FileInputStream(conf));
-            }
-
-            logger.info("## start the canal server.");
-            final CanalController controller = new CanalController(properties);
-            controller.start();
-            running = true;
-            logger.info("## the canal server is running now ......");
-            Runtime.getRuntime().addShutdownHook(new Thread() {
-
-                public void run() {
-                    try {
-                        logger.info("## stop the canal server");
-                        running = false;
-                        controller.stop();
-                    } catch (Throwable e) {
-                        logger.warn("##something goes wrong when stopping canal Server:", e);
-                    } finally {
-                        logger.info("## canal server is down.");
-                    }
-                }
-
-            });
-        } catch (Throwable e) {
-            logger.error("## Something goes wrong when starting up the canal Server:", e);
-            System.exit(0);
-        }
-    }
-
-    public static boolean isRunning() {
-        return running;
-    }
-
-    private static void setGlobalUncaughtExceptionHandler() {
-        Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
-
-            @Override
-            public void uncaughtException(Thread t, Throwable e) {
-                logger.error("UnCaughtException", e);
-            }
-        });
-    }
-}

kafka/src/main/resources/logback.xml

@@ -1,85 +0,0 @@
-<configuration scan="true" scanPeriod=" 5 seconds">
-	<jmxConfigurator />
-	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
-		<encoder>
-			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
-			</pattern>
-		</encoder>
-	</appender>
-	
-	<appender name="CANAL-ROOT" class="ch.qos.logback.classic.sift.SiftingAppender">
-		<discriminator>
-			<Key>destination</Key>
-			<DefaultValue>canal</DefaultValue>
-		</discriminator>
-		<sift>
-			<appender name="FILE-${destination}" class="ch.qos.logback.core.rolling.RollingFileAppender">
-				<File>../logs/${destination}/${destination}.log</File>
-				<rollingPolicy
-					class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-					<!-- rollover daily -->
-					<fileNamePattern>../logs/${destination}/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
-					<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-						<!-- or whenever the file size reaches 100MB -->
-						<maxFileSize>512MB</maxFileSize>
-					</timeBasedFileNamingAndTriggeringPolicy>
-					<maxHistory>60</maxHistory>
-				</rollingPolicy>
-				<encoder>
-					<pattern>
-						%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
-					</pattern>
-				</encoder>
-			</appender>
-		</sift>
-	</appender>
-	
-	<appender name="CANAL-META" class="ch.qos.logback.classic.sift.SiftingAppender">
-		<discriminator>
-			<Key>destination</Key>
-			<DefaultValue>canal</DefaultValue>
-		</discriminator>
-		<sift>
-			<appender name="META-FILE-${destination}" class="ch.qos.logback.core.rolling.RollingFileAppender">
-				<File>../logs/${destination}/meta.log</File>
-				<rollingPolicy
-					class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-					<!-- rollover daily -->
-					<fileNamePattern>../logs/${destination}/%d{yyyy-MM-dd}/meta-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
-					<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-						<!-- or whenever the file size reaches 100MB -->
-						<maxFileSize>32MB</maxFileSize>
-					</timeBasedFileNamingAndTriggeringPolicy>
-					<maxHistory>60</maxHistory>
-				</rollingPolicy>
-				<encoder>
-					<pattern>
-						%d{yyyy-MM-dd HH:mm:ss.SSS} - %msg%n
-					</pattern>
-				</encoder>
-			</appender>
-		</sift>
-	</appender>
-	
-    <logger name="com.alibaba.otter.canal.instance" additivity="false">  
-        <level value="INFO" />  
-        <appender-ref ref="CANAL-ROOT" />
-    </logger>
-    <logger name="com.alibaba.otter.canal.deployer" additivity="false">  
-        <level value="INFO" />  
-        <appender-ref ref="CANAL-ROOT" />
-    </logger>
-    <logger name="com.alibaba.otter.canal.meta.FileMixedMetaManager" additivity="false">  
-        <level value="INFO" />
-        <appender-ref ref="CANAL-META" />
-    </logger>
-	<logger name="com.alibaba.otter.canal.kafka" additivity="false">
-		<level value="INFO" />
-		<appender-ref ref="CANAL-ROOT" />
-	</logger>
-    
-	<root level="WARN">
-		<appender-ref ref="STDOUT"/>
-		<appender-ref ref="CANAL-ROOT" />
-	</root>
-</configuration>

pom.xml

@@ -117,8 +117,6 @@
         <module>client</module>
         <module>deployer</module>
         <module>example</module>
-        <module>kafka</module>
-        <module>kafka-client</module>
         <module>prometheus</module>
         <module>client-adapter</module>
         <module>client-launcher</module>

server/pom.xml

@@ -25,6 +25,29 @@
 			<artifactId>canal.instance.manager</artifactId>
 			<version>${project.version}</version>
 		</dependency>
+		<!-- kafka -->
+		<dependency>
+			<groupId>org.yaml</groupId>
+			<artifactId>snakeyaml</artifactId>
+			<version>1.17</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.11</artifactId>
+			<version>1.1.1</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>slf4j-log4j12</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<!--kafka_2.11_1.1.1 exclusion掉了netty 的依赖，但CanalServerWithNetty 依赖 netty3,升级kafka至 1.1.1 需要显示加入,否则会启动失败 -->
+		<dependency>
+			<groupId>org.jboss.netty</groupId>
+			<artifactId>netty</artifactId>
+			<version>3.2.2.Final</version>
+		</dependency>
 		
 		<!-- test dependency -->
 		<dependency>

kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java → server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.producer;
+package com.alibaba.otter.canal.kafka;
 
 import java.io.IOException;
 import java.util.Properties;
@@ -10,7 +10,6 @@ import org.apache.kafka.common.serialization.StringSerializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**
@@ -21,7 +20,7 @@ import com.alibaba.otter.canal.protocol.Message;
  */
 public class CanalKafkaProducer {
 
-    private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class);
+    private static final Logger       logger = LoggerFactory.getLogger(CanalKafkaProducer.class);
 
     private Producer<String, Message> producer;
 
@@ -49,7 +48,7 @@ public class CanalKafkaProducer {
         }
     }
 
-    public void send(Topic topic, Message message) throws IOException {
+    public void send(KafkaProperties.Topic topic, Message message) throws IOException {
         // set canal.instance.filter.transaction.entry = true
 
         // boolean valid = false;

kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java → server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.producer;
+package com.alibaba.otter.canal.kafka;
 
 import java.io.FileInputStream;
 import java.util.List;
@@ -10,11 +10,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
 
-import com.alibaba.otter.canal.kafka.CanalServerStarter;
-import com.alibaba.otter.canal.kafka.producer.KafkaProperties.CanalDestination;
-import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic;
+import com.alibaba.otter.canal.kafka.KafkaProperties.CanalDestination;
+import com.alibaba.otter.canal.kafka.KafkaProperties.Topic;
 import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.server.CanalServerStarter;
 import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
 
 /**
@@ -23,20 +23,21 @@ import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
  * @author machengyuan 2018-6-11 下午05:30:49
  * @version 1.0.0
  */
-public class CanalKafkaStarter {
+public class CanalKafkaStarter implements CanalServerStarter {
 
-    private static final String       CLASSPATH_URL_PREFIX = "classpath:";
-    private static final Logger       logger               = LoggerFactory.getLogger(CanalKafkaStarter.class);
+    private static final Logger logger               = LoggerFactory.getLogger(CanalKafkaStarter.class);
 
-    private volatile static boolean   running              = false;
+    private static final String CLASSPATH_URL_PREFIX = "classpath:";
 
-    private static ExecutorService    executorService;
+    private volatile boolean    running              = false;
 
-    private static CanalKafkaProducer canalKafkaProducer;
+    private ExecutorService     executorService;
 
-    private static KafkaProperties    kafkaProperties;
+    private CanalKafkaProducer canalKafkaProducer;
 
-    public static void init() {
+    private KafkaProperties kafkaProperties;
+
+    public void init() {
         try {
             logger.info("## load kafka configurations");
             String conf = System.getProperty("kafka.conf", "classpath:kafka.yml");
@@ -96,11 +97,9 @@ public class CanalKafkaStarter {
         }
     }
 
-    private static void worker(CanalDestination destination) {
+    private void worker(CanalDestination destination) {
         while (!running)
             ;
-        while (!CanalServerStarter.isRunning())
-            ; // 等待server启动完成
         logger.info("## start the canal consumer: {}.", destination.getCanalDestination());
         CanalServerWithEmbedded server = CanalServerWithEmbedded.instance();
         ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, "");
@@ -121,7 +120,7 @@ public class CanalKafkaStarter {
                     Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据
                     long batchId = message.getId();
                     try {
-                        int size = message.isRaw() ?  message.getRawEntries().size() : message.getEntries().size();
+                        int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
                         if (batchId != -1 && size != 0) {
                             if (!StringUtils.isEmpty(destination.getTopic())) {
                                 Topic topic = new Topic();

kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java → server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.producer;
+package com.alibaba.otter.canal.kafka;
 
 import java.util.ArrayList;
 import java.util.HashSet;

kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java → server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java

@@ -1,4 +1,4 @@
-package com.alibaba.otter.canal.kafka.producer;
+package com.alibaba.otter.canal.kafka;
 
 import java.util.List;
 import java.util.Map;
@@ -37,20 +37,20 @@ public class MessageSerializer implements Serializer<Message> {
                         List<ByteString> rowEntries = data.getRawEntries();
                         // message size
                         int messageSize = 0;
-                        messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, data.getId());
+                        messageSize += CodedOutputStream.computeInt64Size(1, data.getId());
 
                         int dataSize = 0;
                         for (int i = 0; i < rowEntries.size(); i++) {
-                            dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i));
+                            dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i));
                         }
                         messageSize += dataSize;
                         messageSize += 1 * rowEntries.size();
                         // packet size
                         int size = 0;
-                        size += com.google.protobuf.CodedOutputStream.computeEnumSize(3,
+                        size += CodedOutputStream.computeEnumSize(3,
                             PacketType.MESSAGES.getNumber());
-                        size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
-                                + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
+                        size += CodedOutputStream.computeTagSize(5)
+                                + CodedOutputStream.computeRawVarint32Size(messageSize)
                                 + messageSize;
                         // build data
                         byte[] body = new byte[size];
@@ -73,7 +73,7 @@ public class MessageSerializer implements Serializer<Message> {
                         }
 
                         CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
-                        packetBuilder.setType(CanalPacket.PacketType.MESSAGES);
+                        packetBuilder.setType(PacketType.MESSAGES);
                         packetBuilder.setBody(messageBuilder.build().toByteString());
                         return packetBuilder.build().toByteArray();
                     }

server/src/main/java/com/alibaba/otter/canal/server/CanalServerStarter.java

@@ -0,0 +1,12 @@
+package com.alibaba.otter.canal.server;
+
+/**
+ * 外部服务如Kafka, RocketMQ启动接口
+ *
+ * @author machengyuan 2018-8-23 下午05:20:29
+ * @version 1.0.0
+ */
+public interface CanalServerStarter {
+
+    void init();
+}