diff --git a/common/src/main/java/com/alibaba/otter/canal/common/utils/NamedThreadFactory.java b/common/src/main/java/com/alibaba/otter/canal/common/utils/NamedThreadFactory.java index 866664a1..eeb94c4a 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/utils/NamedThreadFactory.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/utils/NamedThreadFactory.java @@ -23,6 +23,11 @@ public class NamedThreadFactory implements ThreadFactory { public void uncaughtException(Thread t, Throwable e) { + if (e instanceof InterruptedException + || (e.getCause() != null && e.getCause() instanceof InterruptedException)) { + return; + } + logger.error("from " + t.getName(), e); } }; 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 6e02e22d..88278d84 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 @@ -18,7 +18,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.BlockingWaitStrategy; import com.lmax.disruptor.EventFactory; import com.lmax.disruptor.EventHandler; -import com.lmax.disruptor.FatalExceptionHandler; +import com.lmax.disruptor.ExceptionHandler; import com.lmax.disruptor.LifecycleAware; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.Sequence; @@ -85,11 +85,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement this.stageExecutor = Executors.newFixedThreadPool(2, new NamedThreadFactory("MultiStageCoprocessor-other-" + destination)); SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier(); - + ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler(); // stage 2 BatchEventProcessor simpleParserStage = new BatchEventProcessor(disruptorMsgBuffer, sequenceBarrier, new SimpleParserStage()); + simpleParserStage.setExceptionHandler(exceptionHandler); disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence()); // stage 3 @@ -100,7 +101,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement } WorkerPool workerPool = new WorkerPool(disruptorMsgBuffer, dmlParserSequenceBarrier, - new FatalExceptionHandler(), + exceptionHandler, workHandlers); Sequence[] sequence = workerPool.getWorkerSequences(); disruptorMsgBuffer.addGatingSequences(sequence); @@ -110,6 +111,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement BatchEventProcessor sinkStoreStage = new BatchEventProcessor(disruptorMsgBuffer, sinkSequenceBarrier, new SinkStoreStage()); + sinkStoreStage.setExceptionHandler(exceptionHandler); disruptorMsgBuffer.addGatingSequences(sinkStoreStage.getSequence()); // start work @@ -161,7 +163,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void reset() { - stop(); + if (isStart()) { + stop(); + } + start(); } @@ -231,7 +236,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - stop(); + if (isStart()) { + stop(); + } } } @@ -267,7 +274,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - stop(); + if (isStart()) { + stop(); + } } } @@ -306,7 +315,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void onShutdown() { - stop(); + if (isStart()) { + stop(); + } } } @@ -369,6 +380,22 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement } + class SimpleFatalExceptionHandler implements ExceptionHandler { + + @Override + public void handleEventException(final Throwable ex, final long sequence, final Object event) { + throw new RuntimeException(ex); + } + + @Override + public void handleOnStartException(final Throwable ex) { + } + + @Override + public void handleOnShutdownException(final Throwable ex) { + } + } + class MessageEventFactory implements EventFactory { public MessageEvent newInstance() {