fixed test example

This commit is contained in:
agapple
2019-09-02 13:46:35 +08:00
parent 5addd246ac
commit 95050ff382
8 changed files with 75 additions and 67 deletions
@@ -46,7 +46,7 @@ public class CanalLauncher {
properties.load(new FileInputStream(conf));
}
final CanalStater canalStater = new CanalStater(properties);
final CanalStarter canalStater = new CanalStarter(properties);
String managerAddress = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
if (StringUtils.isNotEmpty(managerAddress)) {
String user = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
@@ -20,9 +20,9 @@ import com.alibaba.otter.canal.spi.CanalMQProducer;
* @author rewerma 2018-12-30 下午05:12:16
* @version 1.0.1
*/
public class CanalStater {
public class CanalStarter {
private static final Logger logger = LoggerFactory.getLogger(CanalStater.class);
private static final Logger logger = LoggerFactory.getLogger(CanalStarter.class);
private CanalController controller = null;
private CanalMQProducer canalMQProducer = null;
@@ -33,7 +33,7 @@ public class CanalStater {
private CanalAdminWithNetty canalAdmin;
public CanalStater(Properties properties){
public CanalStarter(Properties properties){
this.properties = properties;
}
@@ -15,7 +15,7 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.admin.CanalAdmin;
import com.alibaba.otter.canal.common.utils.FileUtils;
import com.alibaba.otter.canal.deployer.CanalStater;
import com.alibaba.otter.canal.deployer.CanalStarter;
import com.alibaba.otter.canal.deployer.InstanceConfig;
import com.alibaba.otter.canal.deployer.monitor.InstanceAction;
import com.alibaba.otter.canal.deployer.monitor.InstanceConfigMonitor;
@@ -37,9 +37,9 @@ public class CanalAdminController implements CanalAdmin {
private static final Logger logger = LoggerFactory.getLogger(CanalAdminController.class);
private String user;
private String passwd;
private CanalStater canalStater;
private CanalStarter canalStater;
public CanalAdminController(CanalStater canalStater){
public CanalAdminController(CanalStarter canalStater){
this.canalStater = canalStater;
}
@@ -249,7 +249,7 @@ public class CanalAdminController implements CanalAdmin {
this.passwd = passwd;
}
public void setCanalStater(CanalStater canalStater) {
public void setCanalStater(CanalStarter canalStater) {
this.canalStater = canalStater;
}
+31 -26
View File
@@ -50,31 +50,6 @@ function getMyIp() {
echo $myip
}
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
;;
*)
NET_MODE="--net=host"
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
DATA="$BASE/data"
mkdir -p $DATA
CONFIG=${@:1}
#VOLUMNS="-v $DATA:/home/admin/canal-server/logs"
PORTLIST="11110 11111 11112 9100"
@@ -90,8 +65,38 @@ for PORT in $PORTLIST ; do
fi
done
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
PORTS=""
;;
*)
bin_abs_path=`cd $(dirname $0); pwd`
NET_MODE="--net=host"
PORTS=""
;;
esac
BASE=${bin_abs_path}
DATA="$BASE/data"
mkdir -p $DATA
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
MEMORY="-m 4096m"
LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
echo $cmd
eval $cmd
eval $cmd
+29 -28
View File
@@ -6,9 +6,7 @@ function usage() {
echo "example :"
echo " run_admin.sh -e server.port=8089 \\"
echo " -e canal.adminUser=admin \\"
echo " -e canal.adminPasswd=admin \\"
echo " -e spring.datasource.username=canal \\"
echo " -e spring.datasource.password=canal"
echo " -e canal.adminPasswd=admin"
exit
}
@@ -43,31 +41,6 @@ function getMyIp() {
echo $myip
}
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
;;
*)
NET_MODE="--net=host"
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
DATA="$BASE/data"
mkdir -p $DATA
CONFIG=${@:1}
#VOLUMNS="-v $DATA:/home/admin/canal-admin/logs"
PORTLIST="8089"
@@ -83,6 +56,34 @@ for PORT in $PORTLIST ; do
fi
done
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
PORTS=""
;;
*)
NET_MODE="--net=host"
PORTS=""
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
DATA="$BASE/data"
mkdir -p $DATA
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
MEMORY="-m 1024m"
LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
@@ -74,8 +74,10 @@ public class AbstractCanalClientTest extends BaseCanalClientTest {
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
if (batchId != -1) {
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
}
} catch (Exception e) {
logger.error("process error!", e);
@@ -26,7 +26,7 @@ public class ClusterCanalClientTest extends AbstractCanalClientTest {
// "stability_test", "", "");
// 基于zookeeper动态获取canal server的地址,建立链接,其中一台server发生crash,可以支持failover
CanalConnector connector = CanalConnectors.newClusterConnector("127.0.0.1:2181", destination, "", "");
CanalConnector connector = CanalConnectors.newClusterConnector("127.0.0.1:2181", destination, "canal", "canal");
final ClusterCanalClientTest clientTest = new ClusterCanalClientTest(destination);
clientTest.setConnector(connector);
@@ -24,8 +24,8 @@ public class SimpleCanalClientPermanceTest {
try {
final CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111),
destination,
"",
"");
"canal",
"canal");
Thread ackThread = new Thread(new Runnable() {