From 72b907a75f78b3adae78dba42f16b5c1677cdab9 Mon Sep 17 00:00:00 2001
From: Chuanyi Li L <chuanyi.l.li@ericsson.com>
Date: Thu, 26 Jul 2018 14:30:42 +0800
Subject: [PATCH] Further improvement for issue #771.

---
 .../mysql/MysqlMultiStageCoprocessor.java     | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

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 fb3539a6..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
@@ -64,6 +64,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
     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){
@@ -78,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());
@@ -148,13 +150,17 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
 
     public boolean publish(LogBuffer buffer, String binlogFileName) {
         if (!isStart()) {
+            if (exception != null) {
+                throw exception;
+            }
             return false;
         }
-        /* 由于改为processor退出不stop，那么需要由exception标识coprocessor是否正常工作。
+
+        /* 由于改为processor仅终止自身stage而不是stop，那么需要由incident标识coprocessor是否正常工作。
          * 让dump线程能够及时感知
          */
-        if (exception != null) {
-            throw exception;
+        if (incident != null) {
+            throw incident;
         }
 
         boolean interupted = false;
@@ -399,11 +405,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
 
         @Override
         public void handleEventException(final Throwable ex, final long sequence, final Object event) {
-            /*如果仅仅抛出去，onShutdown会调用stop()，与instance主parse线程的reset()产生竞争，导致向terminated ExecutorService提交新task
+            /*
+             * 若异常不是onEvent body里产生的。如果仅仅抛出去，onShutdown会调用stop()，与dump线程的reset()产生竞争，导致向terminated ExecutorService提交新task
              * #771场景中的异常是InterruptedException，由shutdownNow产生。
              */
-            exception = new CanalParseException(ex);
-            throw exception;
+            incident = new RuntimeException(ex);
+            throw incident;
         }
 
         @Override