Merge pull request #39 from alibaba/master

merge

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

@@ -40,6 +40,22 @@ if [ -z "$JAVA" ]; then
   fi
 fi
 
+case "$#" 
+in
+0 ) 
+  ;;
+2 ) 
+  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
+  ;;
+* )
+  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"
@@ -60,7 +76,7 @@ echo "cd to $bin_abs_path for workaround relative path"
 cd $bin_abs_path
 
 echo CLASSPATH :$CLASSPATH
-$JAVA $JAVA_OPTS $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>$base/logs/adapter.log 2>&1 &
+$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>$base/logs/adapter.log 2>&1 &
 echo $! > $base/bin/adapter.pid
 
 echo "cd to $current_path for continue"

client-adapter/launcher/src/main/resources/application.yml

@@ -7,6 +7,7 @@ spring:
     default-property-inclusion: non_null
 
 canal.conf:
+  mode: tcp # kafka rocketMQ
   canalServerHost: 127.0.0.1:11111
 #  zookeeperHosts: slave1:2181
 #  mqServers: 127.0.0.1:9092 #or rocketmq
@@ -17,7 +18,6 @@ canal.conf:
   timeout:
   accessKey:
   secretKey:
-  mode: tcp # kafka rocketMQ
 #  srcDataSources:
 #    defaultDS:
 #      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true

deployer/src/main/resources/example/instance.properties

@@ -49,6 +49,6 @@ canal.mq.topic=example
 #canal.mq.dynamicTopic=.*,mytest,mytest\\..*,mytest2.user
 canal.mq.partition=0
 # hash partition config
-#canal.mq.partitionsNum=3
+#canal.mq.partitionsNum=4
 #canal.mq.partitionHash=test.table:id^name,.*\\..*
 #################################################

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

@@ -126,7 +126,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
                 producerTmp.commitTransaction();
             }
             callback.commit();
-        } catch (Exception e) {
+        } catch (Throwable e) {
             logger.error(e.getMessage(), e);
             if (kafkaProperties.getTransaction()) {
                 producerTmp.abortTransaction();

server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java

@@ -75,7 +75,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
                 send(destination, destination.getTopic(), data);
             }
             callback.commit();
-        } catch (Exception e) {
+        } catch (Throwable e) {
             callback.rollback();
         }
     }