From 40f8c931c7540c826b9dfc8208e2e5950f0b5dce Mon Sep 17 00:00:00 2001
From: winger <winger2049@gmail.com>
Date: Tue, 25 Sep 2018 18:46:10 +0800
Subject: [PATCH] =?UTF-8?q?fix=20bug=20#968:=20=E5=AF=B9=E4=BA=8Eworkpool?=
 =?UTF-8?q?=E4=B8=8EBatchEventProcessor=E9=9C=80=E8=A6=81=E6=89=8B?=
 =?UTF-8?q?=E5=8A=A8=E8=B0=83=E7=94=A8halt()=EF=BC=8C=E5=90=A6=E5=88=99?=
 =?UTF-8?q?=E4=BC=9A=E5=AF=BC=E8=87=B4=E5=85=B6=E4=BD=BF=E7=94=A8=E7=9A=84?=
 =?UTF-8?q?=E5=B7=A5=E4=BD=9C=E7=BA=BF=E7=A8=8B(=E6=B1=A0)=E6=97=A0?=
 =?UTF-8?q?=E6=B3=95=E6=AD=A3=E5=B8=B8=E5=85=B3=E9=97=AD(parserExecutor/st?=
 =?UTF-8?q?ageExecutor.shutdownNow()=E6=97=A0=E6=95=88)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../mysql/MysqlMultiStageCoprocessor.java     | 21 ++++++++++++++++---
 1 file changed, 18 insertions(+), 3 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 99bb94f5..2e200e2f 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
@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.LockSupport;
 
@@ -36,6 +37,7 @@ import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
 
+
 /**
  * 针对解析器提供一个多阶段协同的处理
  * 
@@ -65,6 +67,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
     private volatile CanalParseException exception;
     private AtomicLong                   eventsPublishBlockingTime;
     private GTIDSet                      gtidSet;
+    private WorkerPool<MessageEvent>     workerPool;
+    private BatchEventProcessor<MessageEvent> simpleParserStage;
+    private BatchEventProcessor<MessageEvent> sinkStoreStage;
 
     public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
                                       EventTransactionBuffer transactionBuffer, String destination){
@@ -91,7 +96,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier();
         ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
         // stage 2
-        BatchEventProcessor<MessageEvent> simpleParserStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
+        simpleParserStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
             sequenceBarrier,
             new SimpleParserStage());
         simpleParserStage.setExceptionHandler(exceptionHandler);
@@ -103,7 +108,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         for (int i = 0; i < parserThreadCount; i++) {
             workHandlers[i] = new DmlParserStage();
         }
-        WorkerPool<MessageEvent> workerPool = new WorkerPool<MessageEvent>(disruptorMsgBuffer,
+        workerPool = new WorkerPool<MessageEvent>(disruptorMsgBuffer,
             dmlParserSequenceBarrier,
             exceptionHandler,
             workHandlers);
@@ -112,7 +117,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
 
         // stage 4
         SequenceBarrier sinkSequenceBarrier = disruptorMsgBuffer.newBarrier(sequence);
-        BatchEventProcessor<MessageEvent> sinkStoreStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
+        sinkStoreStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
             sinkSequenceBarrier,
             new SinkStoreStage());
         sinkStoreStage.setExceptionHandler(exceptionHandler);
@@ -126,14 +131,24 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
 
     @Override
     public void stop() {
+        // fix bug #968，对于pool与
+        workerPool.halt();
+        simpleParserStage.halt();
+        sinkStoreStage.halt();
         try {
             parserExecutor.shutdownNow();
+            while (!parserExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                parserExecutor.shutdownNow();
+            }
         } catch (Throwable e) {
             // ignore
         }
 
         try {
             stageExecutor.shutdownNow();
+            while (!stageExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                stageExecutor.shutdownNow();
+            }
         } catch (Throwable e) {
             // ignore
         }