chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

解决分布式同步开关关闭同步状态下,

Browse Source 
重启 kafka canal-adapter后在开启同步,丢失数据的bug
This commit is contained in:
mcy
2018-10-28 22:35:32 +08:00
parent 00fce30072
commit 167858cd95
5 changed files with 16 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,7 +35,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
}
@Override
protected void closeConnection(){
protected void closeConnection() {
connector.stopRunning();
}
@@ -45,6 +45,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
;
while (running) {
try {
syncSwitch.get(canalDestination);
logger.info("=============> Start to connect topic: {} <=============", this.topic);
connector.connect();
logger.info("=============> Start to subscribe topic: {} <=============", this.topic);
@@ -52,7 +53,11 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
while (running) {
try {
syncSwitch.get(canalDestination);
Boolean status = syncSwitch.status(canalDestination);
if (status != null && !status) {
connector.disconnect();
break;
}
List<?> messages;
if (!flatMessage) {

7
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
View File

@@ -43,6 +43,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
;
while (running) {
try {
syncSwitch.get(canalDestination);
logger.info("=============> Start to connect topic: {} <=============", this.topic);
connector.connect();
logger.info("=============> Start to subscribe topic: {}<=============", this.topic);
@@ -50,6 +51,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
while (running) {
try {
Boolean status = syncSwitch.status(canalDestination);
if (status != null && !status) {
connector.disconnect();
break;
}
List<?> messages;
if (!flatMessage) {
messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS);