From fa8588f766630ecd2b17de5d459ded94bf1dd914 Mon Sep 17 00:00:00 2001
From: duhengforever <duhengforever@gmail.com>
Date: Sun, 30 Sep 2018 09:40:05 +0800
Subject: [PATCH] Modify canal-client config and polish consumer implementation

---
 .../adapter/loader/CanalAdapterRocketMQWorker.java    |  1 +
 client-launcher/src/main/resources/canal-client.yml   |  6 +++---
 .../canal/client/rocketmq/RocketMQCanalConnector.java | 11 ++++++-----
 3 files changed, 10 insertions(+), 8 deletions(-)

diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
index 8c962e8d..98f08286 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
@@ -26,6 +26,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
 
     public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId,
         List<List<CanalOuterAdapter>> canalOuterAdapters) {
+        logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
         this.canalOuterAdapters = canalOuterAdapters;
         this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
         this.topic = topic;
diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml
index 6aac0978..d2cb2175 100644
--- a/client-launcher/src/main/resources/canal-client.yml
+++ b/client-launcher/src/main/resources/canal-client.yml
@@ -1,6 +1,6 @@
 #canalServerHost: 127.0.0.1:11111
 #zookeeperHosts: slave1:2181
-bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers
+bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876
 flatMessage: true
 
 #canalInstances:
@@ -14,9 +14,9 @@ flatMessage: true
 
 mqTopics:
 - mqMode: rocketmq
-- topic: example
+  topic: example
   groups:
-  - groupId: example_g1
+  - groupId: example
     outAdapters:
     - name: logger
 #    - name: hbase
diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
index 0465ec6c..f24955c7 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
+++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
@@ -43,6 +43,9 @@ public class RocketMQCanalConnector implements CanalConnector {
     @Override
     public void connect() throws CanalClientException {
         rocketMQConsumer = new DefaultMQPushConsumer(groupName);
+        if (!StringUtils.isBlank(nameServer)) {
+            rocketMQConsumer.setNamesrvAddr(nameServer);
+        }
     }
 
     @Override
@@ -61,18 +64,16 @@ public class RocketMQCanalConnector implements CanalConnector {
             return;
         }
         try {
-            rocketMQConsumer.subscribe(topic, "*");
             if (rocketMQConsumer == null) {
-                rocketMQConsumer = new DefaultMQPushConsumer(groupName);
-                if (!StringUtils.isBlank(nameServer)) {
-                    rocketMQConsumer.setNamesrvAddr(nameServer);
-                }
+                this.connect();
             }
+            rocketMQConsumer.subscribe(topic, "*");
             rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {
                 @Override
                 public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts,
                     ConsumeOrderlyContext context) {
                     context.setAutoCommit(true);
+                    logger.info("xxxx");
                     boolean isSuccess = process(messageExts);
                     if (isSuccess) {
                         logger.info("Dispatch success!");