From 7ea1468a8d6b05d233e6a84407467c76a6907147 Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Fri, 28 Dec 2018 17:30:04 +0800
Subject: [PATCH] =?UTF-8?q?adapter=20=E5=88=86=E7=BB=84=E9=97=B4=E7=BA=BF?=
 =?UTF-8?q?=E7=A8=8B=E7=BB=9F=E4=B8=80=E6=8A=9B=E5=BC=82=E5=B8=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../loader/AbstractCanalAdapterWorker.java    | 34 +++++++++++--------
 1 file changed, 20 insertions(+), 14 deletions(-)

diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
index b76b9687..0d35fad7 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
@@ -73,16 +73,19 @@ public abstract class AbstractCanalAdapterWorker {
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            futures.forEach(future -> {
+            RuntimeException exception = null;
+            for (Future<Boolean> future : futures) {
                 try {
                     if (!future.get()) {
-                        throw new RuntimeException("Outer adapter sync failed! ");
+                        exception = new RuntimeException("Outer adapter sync failed! ");
                     }
                 } catch (Exception e) {
-                    future.cancel(true);
-                    throw new RuntimeException(e);
+                    exception = new RuntimeException(e);
                 }
-            });
+            }
+            if (exception != null) {
+                throw exception;
+            }
         });
     }
 
@@ -113,22 +116,25 @@ public abstract class AbstractCanalAdapterWorker {
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            futures.forEach(future -> {
+            RuntimeException exception = null;
+            for (Future<Boolean> future : futures) {
                 try {
                     if (!future.get()) {
-                        throw new RuntimeException("Outer adapter sync failed! ");
+                        exception = new RuntimeException("Outer adapter sync failed! ");
                     }
                 } catch (Exception e) {
-                    future.cancel(true);
-                    throw new RuntimeException(e);
+                    exception = new RuntimeException(e);
                 }
-            });
+            }
+            if (exception != null) {
+                throw exception;
+            }
         });
     }
 
     @SuppressWarnings("unchecked")
-    protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage, CanalMQConnector connector,
-                                  ExecutorService workerExecutor) {
+    protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage,
+                                     CanalMQConnector connector, ExecutorService workerExecutor) {
         try {
             List<?> messages;
             if (!flatMessage) {
@@ -157,7 +163,7 @@ public abstract class AbstractCanalAdapterWorker {
                 }
                 connector.ack();
             }
-           return true;
+            return true;
         } catch (Throwable e) {
             if (i == retry - 1) {
                 connector.ack();
@@ -173,7 +179,7 @@ public abstract class AbstractCanalAdapterWorker {
                 // ignore
             }
         }
-        return  false;
+        return false;
     }
 
     /**