From 7c44e5933d2cc62d40e61a316f2b41fab1493dbe Mon Sep 17 00:00:00 2001
From: wallellen <w@741085209630>
Date: Fri, 12 Oct 2018 18:27:32 +0800
Subject: [PATCH] =?UTF-8?q?1.=20AbstractEventParser=20=E4=B8=AD=E5=BA=94?=
 =?UTF-8?q?=E8=AF=A5=E5=AF=B9multiStageCoprocess=E6=AF=8F=E6=AC=A1?=
 =?UTF-8?q?=E5=9C=A8while=E6=9C=80=E5=90=8E=E8=BF=9B=E8=A1=8Cstop=E8=80=8C?=
 =?UTF-8?q?=E4=B8=8D=E6=98=AFreset=EF=BC=8C=20reset=E4=BC=9A=E5=9C=A8mysql?=
 =?UTF-8?q?MultiStageCoprocessor=E4=B8=AD=E9=87=8D=E6=96=B0=E5=88=9B?=
 =?UTF-8?q?=E5=BB=BA=E4=B8=A4=E4=B8=AA=E7=BA=BF=E7=A8=8B=E6=B1=A0=EF=BC=8C?=
 =?UTF-8?q?=20=E5=A6=82=E6=9E=9C=E5=9C=A8dump=E5=87=BA=E9=94=99=E7=9A=84?=
 =?UTF-8?q?=E6=83=85=E5=86=B5=E4=B8=8B=EF=BC=8C=20=E4=BC=9A=E6=97=A0?=
 =?UTF-8?q?=E9=99=90=E5=A4=9A=E7=9A=84=E5=88=9B=E5=BB=BAmultiStageCoproces?=
 =?UTF-8?q?s=E8=80=8C=E4=B8=94=E4=B8=8D=E9=87=8A=E6=94=BE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

2. MySqlMultiStageComprocessor中需要判断是否已经shutdown， 实际情况中存在shutdown为true而terminal一直为false的情况
---
 .../otter/canal/parse/inbound/AbstractEventParser.java    | 4 ++--
 .../parse/inbound/mysql/MysqlMultiStageCoprocessor.java   | 8 ++++++++
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
index e1f80d76..dbb40287 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
@@ -317,10 +317,10 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                     eventSink.interrupt();
                     transactionBuffer.reset();// 重置一下缓冲队列，重新记录数据
                     binlogParser.reset();// 重新置位
-                    if (multiStageCoprocessor != null) {
+                    if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) {
                         // 处理 RejectedExecutionException
                         try {
-                            multiStageCoprocessor.reset();
+                            multiStageCoprocessor.stop();
                         } catch (Throwable t) {
                             logger.debug("multi processor rejected:", t);
                         }
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 2e200e2f..23f399c1 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
@@ -138,6 +138,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         try {
             parserExecutor.shutdownNow();
             while (!parserExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                if (parserExecutor.isShutdown() || parserExecutor.isTerminated()) {
+                    break;
+                }
+
                 parserExecutor.shutdownNow();
             }
         } catch (Throwable e) {
@@ -147,6 +151,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         try {
             stageExecutor.shutdownNow();
             while (!stageExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                if (stageExecutor.isShutdown() || stageExecutor.isTerminated()) {
+                    break;
+                }
+
                 stageExecutor.shutdownNow();
             }
         } catch (Throwable e) {