fixed client-launcher deploy

client-adapter/example/pom.xml

@@ -1,149 +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">
-    <parent>
-        <artifactId>canal.client-adapter</artifactId>
-        <groupId>com.alibaba.otter</groupId>
-        <version>1.1.1-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-    <groupId>com.alibaba.otter</groupId>
-    <artifactId>client-adapter.example</artifactId>
-    <packaging>jar</packaging>
-    <name>canal client adapter example module for otter ${project.version}</name>
-
-    <dependencies>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>client-adapter.common</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>canal.client</artifactId>
-            <version>${canal_version}</version>
-        </dependency>
-        <!-- 单独引入rocketmq依赖 -->
-        <dependency>
-            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-client</artifactId>
-            <version>4.3.0</version>
-        </dependency>
-        <!-- 单独引入kafka依赖 -->
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-            <version>1.1.1</version>
-        </dependency>
-        <dependency>
-            <groupId>org.yaml</groupId>
-            <artifactId>snakeyaml</artifactId>
-            <version>1.17</version>
-        </dependency>
-
-        <!-- outer adapter -->
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>client-adapter.logger</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <!-- deploy模块的packaging通常是jar，如果项目中没有java 源代码或资源文件，加上这一段配置使项目能通过构建 -->
-            <plugin>
-                <artifactId>maven-jar-plugin</artifactId>
-                <version>3.0.2</version>
-                <configuration>
-                    <archive>
-                        <addMavenDescriptor>true</addMavenDescriptor>
-                    </archive>
-                    <excludes>
-                        <exclude>**/logback.xml</exclude>
-                        <exclude>**/canal-client.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_client</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>

client-adapter/example/src/main/assembly/dev.xml

@@ -1,57 +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>./src/main/conf</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<excludes>
-				<exclude>META-INF/**</exclude>
-			</excludes>
-		</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>

client-adapter/example/src/main/assembly/release.xml

@@ -1,57 +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>./src/main/conf</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-		</fileSet>
-		<fileSet>
-			<directory>./src/main/resources</directory>
-			<outputDirectory>/conf</outputDirectory>
-			<includes>
-				<include>**/*</include>
-			</includes>
-			<excludes>
-				<exclude>META-INF/**</exclude>
-			</excludes>
-		</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>

client-adapter/example/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-client.yml
-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.client.ClientLauncher
-echo start cmd : %CMD_STR%
-
-java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.client.ClientLauncher

client-adapter/example/src/main/bin/startup.sh

@@ -1,100 +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-client.yml
-logback_configurationFile=$base/conf/logback.xml
-export LANG=en_US.UTF-8
-export BASE=$base
-
-if [ -f $base/bin/canal_client.pid ] ; then
-	echo "found canal_client.pid , Please run stop.sh first ,then startup.sh" 2>&2
-    exit 1
-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-client -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.client.ClientLauncher 1>>$base/logs/canal_client.log 2>&1 &
-	echo $! > $base/bin/canal_client.pid
-	
-	echo "cd to $current_path for continue"
-  	cd $current_path
-else 
-	echo "canal client conf("$canal_conf") OR log configration file($logback_configurationFile) is not exist,please create then first!"
-fi

client-adapter/example/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_client.pid
-if [ ! -f "$pidfile" ];then
-	echo "canal client is not running. exists"
-	exit
-fi
-
-pid=`cat $pidfile`
-if [ "$pid" == "" ] ; then
-	pid=`get_pid "appName=otter-canal-client"`
-fi
-
-echo -e "`hostname`: stopping canal $pid ... "
-kill $pid
-
-LOOPS=0
-while (true); 
-do 
-	gpid=`get_pid "appName=otter-canal-client" "$pid"`
-    if [ "$gpid" == "" ] ; then
-    	echo "Oook! cost:$LOOPS"
-    	`rm $pidfile`
-    	break;
-    fi
-    let LOOPS=LOOPS+1
-    sleep 1
-done

client-adapter/example/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java

@@ -1,66 +0,0 @@
-package com.alibaba.otter.canal.client;
-
-import java.io.FileInputStream;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
-
-import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader;
-import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
-
-public class ClientLauncher {
-
-    private static final String CLASSPATH_URL_PREFIX = "classpath:";
-    private static final Logger logger               = LoggerFactory.getLogger(ClientLauncher.class);
-
-    public static void main(String[] args) {
-        try {
-            logger.info("## set default uncaught exception handler");
-            setGlobalUncaughtExceptionHandler();
-
-            logger.info("## load canal client configurations");
-            String conf = System.getProperty("client.conf", "classpath:canal-client.yml");
-            CanalClientConfig canalClientConfig;
-            if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
-                conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
-                canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf),
-                    CanalClientConfig.class);
-            } else {
-                canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class);
-            }
-            logger.info("## start the canal client adapters.");
-            final CanalAdapterLoader adapterLoader = new CanalAdapterLoader(canalClientConfig);
-            adapterLoader.init();
-            logger.info("## the canal client adapters are running now ......");
-            Runtime.getRuntime().addShutdownHook(new Thread() {
-
-                public void run() {
-                    try {
-                        logger.info("## stop the canal client adapters");
-                        adapterLoader.destroy();
-                    } catch (Throwable e) {
-                        logger.warn("## something goes wrong when stopping canal client adapters:", e);
-                    } finally {
-                        logger.info("## canal client adapters are down.");
-                    }
-                }
-
-            });
-        } catch (Throwable e) {
-            logger.error("## something goes wrong when starting up the canal client adapters:", e);
-            System.exit(0);
-        }
-    }
-
-    private static void setGlobalUncaughtExceptionHandler() {
-        Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
-
-            @Override
-            public void uncaughtException(Thread t, Throwable e) {
-                logger.error("UnCaughtException", e);
-            }
-        });
-    }
-}

client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java

@@ -1,168 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.loader;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
-import com.alibaba.otter.canal.protocol.FlatMessage;
-import com.alibaba.otter.canal.protocol.Message;
-
-/**
- * 适配器工作线程抽象类
- *
- * @author machengyuan 2018-8-19 下午11:30:49
- * @version 1.0.0
- */
-public abstract class AbstractCanalAdapterWorker {
-
-    protected final Logger                    logger  = LoggerFactory.getLogger(this.getClass());
-
-    protected String                          canalDestination;                                                 // canal实例
-    protected List<List<OuterAdapter>>        canalOuterAdapters;                                               // 外部适配器
-    protected ExecutorService                 groupInnerExecutorService;                                        // 组内工作线程池
-    protected volatile boolean                running = false;                                                  // 是否运行中
-    protected Thread                          thread  = null;
-    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
-
-                                                          @Override
-                                                          public void uncaughtException(Thread t, Throwable e) {
-                                                              logger.error("parse events has an error", e);
-                                                          }
-                                                      };
-
-    protected void writeOut(final Message message) {
-        List<Future<Boolean>> futures = new ArrayList<>();
-        // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            final List<OuterAdapter> adapters = outerAdapters;
-            futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
-
-                @Override
-                public Boolean call() {
-                    try {
-                        // 组内适配器穿行运行，尽量不要配置组内适配器
-                        for (final OuterAdapter c : adapters) {
-                            long begin = System.currentTimeMillis();
-                            MessageUtil.parse4Dml(canalDestination, message, new MessageUtil.Consumer<Dml>() {
-
-                                @Override
-                                public void accept(Dml dml) {
-                                    c.sync(dml);
-                                }
-                            });
-
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("{} elapsed time: {}",
-                                    c.getClass().getName(),
-                                    (System.currentTimeMillis() - begin));
-                            }
-                        }
-                        return true;
-                    } catch (Exception e) {
-                        return false;
-                    }
-                }
-            }));
-
-            // 等待所有适配器写入完成
-            // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
-                try {
-                    if (!f.get()) {
-                        logger.error("Outer adapter write failed");
-                    }
-                } catch (InterruptedException | ExecutionException e) {
-                    // ignore
-                }
-            }
-        }
-    }
-
-    protected void writeOut(final FlatMessage flatMessage) {
-        List<Future<Boolean>> futures = new ArrayList<>();
-        // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            final List<OuterAdapter> adapters = outerAdapters;
-            futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
-
-                @Override
-                public Boolean call() {
-                    try {
-                        // 组内适配器穿行运行，尽量不要配置组内适配器
-                        for (OuterAdapter c : adapters) {
-                            long begin = System.currentTimeMillis();
-                            Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
-                            c.sync(dml);
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("{} elapsed time: {}",
-                                    c.getClass().getName(),
-                                    (System.currentTimeMillis() - begin));
-                            }
-                        }
-                        return true;
-                    } catch (Exception e) {
-                        return false;
-                    }
-                }
-            }));
-
-            // 等待所有适配器写入完成
-            // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
-                try {
-                    if (!f.get()) {
-                        logger.error("Outer adapter write failed");
-                    }
-                } catch (InterruptedException | ExecutionException e) {
-                    // ignore
-                }
-            }
-        }
-    }
-
-    protected void writeOut(Message message, String topic) {
-        if (logger.isDebugEnabled()) {
-            logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size());
-        }
-        long begin = System.currentTimeMillis();
-        writeOut(message);
-        long now = System.currentTimeMillis();
-        if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
-            logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
-        }
-        if (logger.isDebugEnabled()) {
-            logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
-        }
-    }
-
-    protected void stopOutAdapters() {
-        if (thread != null) {
-            try {
-                thread.join();
-            } catch (InterruptedException e) {
-                // ignore
-            }
-        }
-        groupInnerExecutorService.shutdown();
-        logger.info("topic connectors' worker thread dead!");
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            for (OuterAdapter adapter : outerAdapters) {
-                adapter.destroy();
-            }
-        }
-        logger.info("topic all connectors destroyed!");
-    }
-
-    public abstract void start();
-
-    public abstract void stop();
-}

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

@@ -1,178 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.loader;
-
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.kafka.clients.consumer.CommitFailedException;
-import org.apache.kafka.common.errors.WakeupException;
-
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
-import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors;
-import com.alibaba.otter.canal.protocol.FlatMessage;
-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;
-
-    private String              topic;
-
-    private boolean             flatMessage;
-
-    public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId,
-                                   List<List<OuterAdapter>> canalOuterAdapters, boolean flatMessage){
-        this.canalOuterAdapters = canalOuterAdapters;
-        this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
-        this.topic = topic;
-        this.canalDestination = topic;
-        this.flatMessage = flatMessage;
-        connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage);
-        // connector.setSessionTimeout(1L, TimeUnit.MINUTES);
-
-        // super.initSwitcher(topic);
-    }
-
-    @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(new Runnable() {
-
-                @Override
-                public void run() {
-                    process();
-                }
-            });
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
-    }
-
-    @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            connector.stopRunning();
-            running = false;
-
-            // if (switcher != null && !switcher.state()) {
-            // switcher.set(true);
-            // }
-
-            if (thread != null) {
-                try {
-                    thread.join();
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-            groupInnerExecutorService.shutdown();
-            logger.info("topic {} connectors' worker thread dead!", this.topic);
-            for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("topic {} all connectors destroyed!", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
-        while (!running)
-            ;
-        ExecutorService executor = Executors.newSingleThreadExecutor();
-        // final AtomicBoolean executing = new AtomicBoolean(true);
-        while (running) {
-            try {
-                logger.info("=============> Start to connect topic: {} <=============", this.topic);
-                connector.connect();
-                logger.info("=============> Start to subscribe topic: {} <=============", this.topic);
-                connector.subscribe();
-                logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
-                while (running) {
-                    try {
-                        // switcher.get(); //等待开关开启
-
-                        List<?> messages;
-                        if (!flatMessage) {
-                            messages = connector.getWithoutAck();
-                        } else {
-                            messages = connector.getFlatMessageWithoutAck(100L, TimeUnit.MILLISECONDS);
-                        }
-                        if (messages != null) {
-                            for (final Object message : messages) {
-                                if (message instanceof FlatMessage) {
-                                    writeOut((FlatMessage) message);
-                                } else {
-                                    writeOut((Message) message);
-                                }
-                                // executing.set(true);
-                                // if (message != null) {
-                                // executor.submit(new Runnable() {
-                                //
-                                // @Override
-                                // public void run() {
-                                // try {
-                                // if (message instanceof FlatMessage) {
-                                // writeOut((FlatMessage) message);
-                                // } else {
-                                // writeOut((Message) message);
-                                // }
-                                // } catch (Exception e) {
-                                // logger.error(e.getMessage(), e);
-                                // } finally {
-                                // executing.compareAndSet(true, false);
-                                // }
-                                // }
-                                // });
-                                //
-                                // // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
-                                // long currentTS = System.currentTimeMillis();
-                                // while (executing.get()) {
-                                // // 大于10秒未消费完ack一次keep alive
-                                // if (System.currentTimeMillis() - currentTS > 10000) {
-                                // connector.ack();
-                                // currentTS = System.currentTimeMillis();
-                                // }
-                                // }
-                                // }
-                            }
-                        }
-                        connector.ack();
-                    } catch (CommitFailedException e) {
-                        logger.warn(e.getMessage());
-                    } catch (Exception e) {
-                        logger.error(e.getMessage(), e);
-                        TimeUnit.SECONDS.sleep(1L);
-                    }
-                }
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-
-        executor.shutdown();
-
-        try {
-            connector.unsubscribe();
-        } catch (WakeupException e) {
-            // No-op. Continue process
-        }
-        connector.disconnect();
-        logger.info("=============> Disconnect topic: {} <=============", this.topic);
-    }
-}

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

@@ -1,177 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.loader;
-
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
-import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
-import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
-
-/**
- * MQ外部适配器的加载器
- *
- * @version 1.0.0
- */
-public class CanalAdapterLoader {
-
-    private static final Logger                     logger        = LoggerFactory.getLogger(CanalAdapterLoader.class);
-
-    private CanalClientConfig                       canalClientConfig;
-
-    private Map<String, CanalAdapterWorker>         canalWorkers  = new HashMap<>();
-
-    private Map<String, AbstractCanalAdapterWorker> canalMQWorker = new HashMap<>();
-
-    private ExtensionLoader<OuterAdapter>      loader;
-
-    public CanalAdapterLoader(CanalClientConfig canalClientConfig){
-        this.canalClientConfig = canalClientConfig;
-    }
-
-    /**
-     * 初始化canal-client、 canal-client-rocketmq的适配器
-     */
-    public void init() {
-        // canal instances 和 mq topics 配置不能同时为空
-        if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) {
-            throw new RuntimeException("Blank config property: canalInstances or canalMQTopics");
-        }
-
-        loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class, "" /*
-                                                                                 * TODO
-                                                                                 * canalClientConfig
-                                                                                 * .
-                                                                                 * getClassloaderPolicy
-                                                                                 * (
-                                                                                 * )
-                                                                                 */);
-
-        String canalServerHost = this.canalClientConfig.getCanalServerHost();
-        SocketAddress sa = null;
-        if (canalServerHost != null) {
-            String[] ipPort = canalServerHost.split(":");
-            sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
-        }
-        String zkHosts = this.canalClientConfig.getZookeeperHosts();
-
-        // 初始化canal-client的适配器
-        if (canalClientConfig.getCanalInstances() != null) {
-            for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) {
-                List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
-
-                for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) {
-                    List<OuterAdapter> canalOutConnectors = new ArrayList<>();
-                    for (OuterAdapterConfig c : connectorGroup.getOutAdapters()) {
-                        loadConnector(c, canalOutConnectors);
-                    }
-                    canalOuterAdapterGroups.add(canalOutConnectors);
-                }
-                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());
-            }
-        }
-
-        // 初始化canal-client-mq的适配器
-        if (canalClientConfig.getMqTopics() != null) {
-            for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) {
-                for (CanalClientConfig.Group group : topic.getGroups()) {
-                    List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
-
-                    List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
-
-                    for (OuterAdapterConfig config : group.getOutAdapters()) {
-                        loadConnector(config, canalOuterAdapters);
-                    }
-                    canalOuterAdapterGroups.add(canalOuterAdapters);
-                    if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) {
-                        CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig.getBootstrapServers(),
-                            topic.getTopic(),
-                            group.getGroupId(),
-                            canalOuterAdapterGroups);
-                        canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
-                        rocketMQWorker.start();
-                    } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) {
-                        CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig.getBootstrapServers(),
-                            topic.getTopic(),
-                            group.getGroupId(),
-                            canalOuterAdapterGroups,
-                            canalClientConfig.getFlatMessage());
-                        canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
-                        canalKafkaWorker.start();
-                    }
-                    logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-"
-                                                                                             + group.getGroupId());
-
-                }
-            }
-        }
-    }
-
-    private void loadConnector(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
-        try {
-            OuterAdapter adapter = loader.getExtension(config.getName());
-            ClassLoader cl = Thread.currentThread().getContextClassLoader();
-            // 替换ClassLoader
-            Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());
-            adapter.init(config);
-            Thread.currentThread().setContextClassLoader(cl);
-            canalOutConnectors.add(adapter);
-            logger.info("Load canal adapter: {} succeed", config.getName());
-        } catch (Exception e) {
-            logger.error("Load canal adapter: {} failed", config.getName(), e);
-        }
-    }
-
-    /**
-     * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
-     */
-    public void destroy() {
-        if (canalWorkers.size() > 0) {
-            ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
-            for (CanalAdapterWorker v : canalWorkers.values()) {
-                final CanalAdapterWorker caw = v;
-                stopExecutorService.submit(new Runnable() {
-
-                    @Override
-                    public void run() {
-                        caw.stop();
-                    }
-                });
-            }
-            stopExecutorService.shutdown();
-        }
-        if (canalMQWorker.size() > 0) {
-            ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size());
-            for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) {
-                final AbstractCanalAdapterWorker worker = tmp;
-                stopMQWokerService.submit(new Runnable() {
-
-                    @Override
-                    public void run() {
-                        worker.stop();
-                    }
-                });
-            }
-            stopMQWokerService.shutdown();
-        }
-        logger.info("All canal adapters destroyed");
-    }
-}

client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java

@@ -1,124 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.loader;
-
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.kafka.clients.consumer.CommitFailedException;
-import org.apache.kafka.common.errors.WakeupException;
-
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
-import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider;
-import com.alibaba.otter.canal.protocol.Message;
-
-/**
- * kafka对应的client适配器工作线程
- *
- * @author machengyuan 2018-8-19 下午11:30:49
- * @version 1.0.0
- */
-public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
-
-    private RocketMQCanalConnector connector;
-
-    private String                 topic;
-
-    public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId,
-                                      List<List<OuterAdapter>> canalOuterAdapters){
-        logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
-        this.canalOuterAdapters = canalOuterAdapters;
-        this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
-        this.topic = topic;
-        this.canalDestination = topic;
-        connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId);
-    }
-
-    @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(new Runnable() {
-
-                @Override
-                public void run() {
-                    process();
-                }
-            });
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
-    }
-
-    @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-            connector.stopRunning();
-            running = false;
-            logger.info("Stop topic {} out adapters begin", this.topic);
-            stopOutAdapters();
-            logger.info("Stop topic {} out adapters end", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
-        while (!running)
-            ;
-        ExecutorService executor = Executors.newSingleThreadExecutor();
-        while (running) {
-            try {
-                logger.info("=============> Start to connect topic: {} <=============", this.topic);
-                connector.connect();
-                logger.info("=============> Start to subscribe topic: {}<=============", this.topic);
-                connector.subscribe();
-                logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
-                while (running) {
-                    try {
-                        // switcher.get(); //等待开关开启
-
-                        final Message message = connector.getWithoutAck(1);
-                        if (message != null) {
-                            executor.submit(new Runnable() {
-
-                                @Override
-                                public void run() {
-                                    try {
-                                        writeOut(message, topic);
-                                    } catch (Exception e) {
-                                        logger.error(e.getMessage(), e);
-                                    }
-                                    connector.ack(message.getId());
-                                }
-                            });
-                        } else {
-                            logger.debug("Message is null");
-                        }
-                    } catch (CommitFailedException e) {
-                        logger.warn(e.getMessage());
-                    } catch (Exception e) {
-                        logger.error(e.getMessage(), e);
-                        TimeUnit.SECONDS.sleep(1L);
-                    }
-                }
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-
-        executor.shutdown();
-
-        try {
-            connector.unsubscribe();
-        } catch (WakeupException e) {
-            // No-op. Continue process
-        }
-        connector.stopRunning();
-        logger.info("=============> Disconnect topic: {} <=============", this.topic);
-    }
-}

client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java

@@ -1,186 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.loader;
-
-import java.net.SocketAddress;
-import java.util.List;
-import java.util.concurrent.Executors;
-
-import com.alibaba.otter.canal.client.CanalConnector;
-import com.alibaba.otter.canal.client.CanalConnectors;
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
-import com.alibaba.otter.canal.protocol.Message;
-
-/**
- * 原生canal-server对应的client适配器工作线程
- *
- * @author machengyuan 2018-8-19 下午11:30:49
- * @version 1.0.0
- */
-public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
-
-    private static final int BATCH_SIZE = 50;
-    private static final int SO_TIMEOUT = 0;
-
-    private CanalConnector   connector;
-
-    /**
-     * 单台client适配器worker的构造方法
-     *
-     * @param canalDestination canal实例名
-     * @param address canal-server地址
-     * @param canalOuterAdapters 外部适配器组
-     */
-    public CanalAdapterWorker(String canalDestination, SocketAddress address,
-                              List<List<OuterAdapter>> canalOuterAdapters){
-        this.canalOuterAdapters = canalOuterAdapters;
-        this.canalDestination = canalDestination;
-        groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
-        connector = CanalConnectors.newSingleConnector(address, canalDestination, "", "");
-    }
-
-    /**
-     * HA模式下client适配器worker的构造方法
-     *
-     * @param canalDestination canal实例名
-     * @param zookeeperHosts zookeeper地址
-     * @param canalOuterAdapters 外部适配器组
-     */
-    public CanalAdapterWorker(String canalDestination, String zookeeperHosts,
-                              List<List<OuterAdapter>> canalOuterAdapters){
-        this.canalOuterAdapters = canalOuterAdapters;
-        this.canalDestination = canalDestination;
-        groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
-        connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");
-        ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT);
-
-        // super.initSwitcher(canalDestination);
-    }
-
-    @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(new Runnable() {
-
-                @Override
-                public void run() {
-                    process();
-                }
-            });
-            thread.setUncaughtExceptionHandler(handler);
-            thread.start();
-            running = true;
-        }
-    }
-
-    @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            // if (switcher != null && !switcher.state()) {
-            // switcher.set(true);
-            // }
-
-            connector.stopRunning();
-            running = false;
-
-            logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination);
-            if (thread != null) {
-                try {
-                    thread.join();
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-            groupInnerExecutorService.shutdown();
-            logger.info("destination {} adapters' worker thread dead!", canalDestination);
-            for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("destination {} all adapters destroyed!", canalDestination);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
-        while (!running)
-            ; // waiting until running == true
-        while (running) {
-            try {
-                // if (switcher != null) {
-                // switcher.get();
-                // }
-                logger.info("=============> Start to connect destination: {} <=============", this.canalDestination);
-                connector.connect();
-                logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination);
-                connector.subscribe();
-                logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination);
-                while (running) {
-                    // try {
-                    // if (switcher != null) {
-                    // switcher.get();
-                    // }
-                    // } catch (TimeoutException e) {
-                    // break;
-                    // }
-
-                    // server配置canal.instance.network.soTimeout(默认: 30s)
-                    // 范围内未与server交互，server将关闭本次socket连接
-                    Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
-                    long batchId = message.getId();
-                    try {
-                        int size = message.getEntries().size();
-
-                        if (batchId == -1 || size == 0) {
-                            try {
-                                Thread.sleep(1000);
-                            } catch (InterruptedException e) {
-                                // ignore
-                            }
-                        } else {
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("destination: {} batchId: {} batchSize: {} ",
-                                    this.canalDestination,
-                                    batchId,
-                                    size);
-                            }
-                            long begin = System.currentTimeMillis();
-                            writeOut(message);
-                            long now = System.currentTimeMillis();
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("destination: {} batchId: {} elapsed time: {} ms",
-                                    this.canalDestination,
-                                    batchId,
-                                    now - begin);
-                            }
-                        }
-                        connector.ack(batchId); // 提交确认
-                    } catch (Exception e) {
-                        connector.rollback(batchId); // 处理失败, 回滚数据
-                        throw e;
-                    }
-                }
-
-            } catch (Exception e) {
-                logger.error("process error!", e);
-            } finally {
-                connector.disconnect();
-                logger.info("=============> Disconnect destination: {} <=============", this.canalDestination);
-            }
-
-            if (running) { // is reconnect
-                try {
-                    Thread.sleep(1000);
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-
-        }
-    }
-}

client-adapter/example/src/main/resources/canal-client.yml

@@ -1,18 +0,0 @@
-canalServerHost: 127.0.0.1:11111
-#zookeeperHosts: slave1:2181
-#bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876
-flatMessage: false
-
-canalInstances:
-- instance: example
-  adapterGroups:
-  - outAdapters:
-    - name: logger
-
-#mqTopics:
-#- mqMode: rocketmq
-#  topic: example
-#  groups:
-#  - groupId: example
-#    outAdapters:
-#    - name: logger

client-adapter/example/src/main/resources/logback.xml

@@ -1,47 +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_client</DefaultValue>
-		</discriminator>
-		<sift>
-			<appender name="FILE-${destination}" class="ch.qos.logback.core.rolling.RollingFileAppender">
-				<File>../logs/${destination}.log</File>
-				<rollingPolicy
-					class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-					<!-- rollover daily -->
-					<fileNamePattern>../logs/%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>
-
-	<logger name="com.alibaba.otter.canal.client" additivity="false">
-		<level value="INFO" />
-		<appender-ref ref="STDOUT"/>
-		<appender-ref ref="CANAL-ROOT" />
-	</logger>
-    
-	<root level="WARN">
-		<!--<appender-ref ref="STDOUT"/>-->
-		<appender-ref ref="CANAL-ROOT" />
-	</root>
-</configuration>

client-adapter/hbase/pom.xml

@@ -43,48 +43,60 @@
         </dependency>
     </dependencies>
 
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <version>2.4</version>
-                <configuration>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-            <plugin>
-                <artifactId>maven-antrun-plugin</artifactId>
-                <executions>
-                    <execution>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>run</goal>
-                        </goals>
+
+    <profiles>
+        <profile>
+            <id>release</id>
+            <activation>
+                <property>
+                    <name>env</name>
+                    <value>release</value>
+                </property>
+            </activation>
+
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-assembly-plugin</artifactId>
+                        <version>2.4</version>
                         <configuration>
-                            <tasks>
-                                <copy todir="${project.basedir}/../launcher/target/config/hbase" overwrite="true" >
-                                    <fileset dir="${project.basedir}/src/main/resources/hbase" erroronmissingdir="true">
-                                        <include name="*.yml"/>
-                                    </fileset>
-                                </copy>
-                            </tasks>
+                            <descriptorRefs>
+                                <descriptorRef>jar-with-dependencies</descriptorRef>
+                            </descriptorRefs>
                         </configuration>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
-
+                        <executions>
+                            <execution>
+                                <id>make-assembly</id>
+                                <phase>package</phase>
+                                <goals>
+                                    <goal>single</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                    <plugin>
+                        <artifactId>maven-antrun-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <phase>package</phase>
+                                <goals>
+                                    <goal>run</goal>
+                                </goals>
+                                <configuration>
+                                    <tasks>
+                                        <copy todir="${project.basedir}/../launcher/target/config/hbase" overwrite="true" >
+                                            <fileset dir="${project.basedir}/src/main/resources/hbase" erroronmissingdir="true">
+                                                <include name="*.yml"/>
+                                            </fileset>
+                                        </copy>
+                                    </tasks>
+                                </configuration>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

client-adapter/launcher/pom.xml

@@ -104,6 +104,18 @@
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
                 <version>2.0.1.RELEASE</version>
+                <configuration>
+                    <excludes>
+                        <exclude>
+                            <groupId>com.alibaba.otter</groupId>
+                            <artifactId>client-adapter.hbase</artifactId>
+                        </exclude>
+                        <exclude>
+                            <groupId>com.alibaba.otter</groupId>
+                            <artifactId>client-adapter.logger</artifactId>
+                        </exclude>
+                    </excludes>
+                </configuration>
                 <executions>
                     <execution>
                         <goals>
@@ -150,24 +162,43 @@
                     </execution>
                 </executions>
             </plugin>
-            <plugin>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <configuration>
-                    <appendAssemblyId>false</appendAssemblyId>
-                    <descriptors>
-                        <descriptor>assembly.xml</descriptor>
-                    </descriptors>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
         </plugins>
     </build>
+
+    <profiles>
+        <profile>
+            <id>release</id>
+            <activation>
+                <property>
+                    <name>env</name>
+                    <value>release</value>
+                </property>
+            </activation>
+
+            <build>
+                <plugins>
+                    <plugin>
+                        <artifactId>maven-assembly-plugin</artifactId>
+                        <configuration>
+                            <appendAssemblyId>false</appendAssemblyId>
+                            <descriptors>
+                                <descriptor>${basedir}/src/main/assembly/release.xml</descriptor>
+                            </descriptors>
+                            <finalName>${project.artifactId}-${project.version}</finalName>
+                            <outputDirectory>${project.basedir}/../../target</outputDirectory>
+                        </configuration>
+                        <executions>
+                            <execution>
+                                <id>make-assembly</id>
+                                <phase>package</phase>
+                                <goals>
+                                    <goal>single</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

client-adapter/logger/pom.xml

@@ -21,27 +21,39 @@
         </dependency>
     </dependencies>
 
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <version>2.4</version>
-                <configuration>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
+    <profiles>
+        <profile>
+            <id>release</id>
+            <activation>
+                <property>
+                    <name>env</name>
+                    <value>release</value>
+                </property>
+            </activation>
+
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-assembly-plugin</artifactId>
+                        <version>2.4</version>
+                        <configuration>
+                            <descriptorRefs>
+                                <descriptorRef>jar-with-dependencies</descriptorRef>
+                            </descriptorRefs>
+                        </configuration>
+                        <executions>
+                            <execution>
+                                <id>make-assembly</id>
+                                <phase>package</phase>
+                                <goals>
+                                    <goal>single</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

client-adapter/pom.xml

@@ -24,7 +24,6 @@
         <module>logger</module>
         <module>hbase</module>
         <module>launcher</module>
-        <module>example</module>
     </modules>
 
     <build>