diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 65904b34..71dc7ca9 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -4,6 +4,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.locks.LockSupport; +import com.alibaba.otter.canal.common.CanalException; import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; @@ -52,17 +53,18 @@ import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; */ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implements MultiStageCoprocessor { - private LogEventConvert logEventConvert; - private EventTransactionBuffer transactionBuffer; - private ErosaConnection connection; + private LogEventConvert logEventConvert; + private EventTransactionBuffer transactionBuffer; + private ErosaConnection connection; - private int parserThreadCount; - private int ringBufferSize; - private RingBuffer disruptorMsgBuffer; - private ExecutorService parserExecutor; - private ExecutorService stageExecutor; - private String destination; - private CanalParseException exception; + private int parserThreadCount; + private int ringBufferSize; + private RingBuffer disruptorMsgBuffer; + private ExecutorService parserExecutor; + private ExecutorService stageExecutor; + private String destination; + private volatile CanalParseException exception; + private volatile RuntimeException incident; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -77,6 +79,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public void start() { super.start(); this.exception = null; + this.incident = null; this.disruptorMsgBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(), ringBufferSize, new BlockingWaitStrategy()); @@ -153,6 +156,13 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement return false; } + /* 由于改为processor仅终止自身stage而不是stop,那么需要由incident标识coprocessor是否正常工作。 + * 让dump线程能够及时感知 + */ + if (incident != null) { + throw incident; + } + boolean interupted = false; do { try { @@ -253,9 +263,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - if (isStart()) { - stop(); - } + } } @@ -291,9 +299,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - if (isStart()) { - stop(); - } + } } @@ -332,9 +338,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - if (isStart()) { - stop(); - } + } } @@ -401,7 +405,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void handleEventException(final Throwable ex, final long sequence, final Object event) { - throw new RuntimeException(ex); + /* + * 若异常不是onEvent body里产生的。如果仅仅抛出去,onShutdown会调用stop(),与dump线程的reset()产生竞争,导致向terminated ExecutorService提交新task + * #771场景中的异常是InterruptedException,由shutdownNow产生。 + */ + incident = new RuntimeException(ex); + throw incident; } @Override