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 48c0e57c..e4d58272 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 @@ -76,16 +76,19 @@ public abstract class AbstractCanalAdapterWorker { // 等待所有适配器写入完成 // 由于是组间并发操作,所以将阻塞直到耗时最久的工作组操作完成 - futures.forEach(future -> { + RuntimeException exception = null; + for (Future 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; + } }); } @@ -116,16 +119,19 @@ public abstract class AbstractCanalAdapterWorker { // 等待所有适配器写入完成 // 由于是组间并发操作,所以将阻塞直到耗时最久的工作组操作完成 - futures.forEach(future -> { + RuntimeException exception = null; + for (Future 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; + } }); }