chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1355 from rewerma/master

Browse Source 
adapter 分组间线程统一抛异常
This commit is contained in:
agapple
2018-12-28 23:06:08 +08:00
committed by GitHub
parent 14ba358525 7ea1468a8d
commit 8d84521767
1 changed files with 16 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
View File

@@ -76,16 +76,19 @@ public abstract class AbstractCanalAdapterWorker {
// 等待所有适配器写入完成 // 等待所有适配器写入完成
// 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
futures.forEach(future -> { RuntimeException exception = null;
for (Future<Boolean> future : futures) {
try { try {
if (!future.get()) { if (!future.get()) {
throw new RuntimeException("Outer adapter sync failed! "); exception = new RuntimeException("Outer adapter sync failed! ");
} }
} catch (Exception e) { } catch (Exception e) {
future.cancel(true); exception = new RuntimeException(e);
throw new RuntimeException(e);
} }
}); }
if (exception != null) {
throw exception;
}
}); });
} }
@@ -116,16 +119,19 @@ public abstract class AbstractCanalAdapterWorker {
// 等待所有适配器写入完成 // 等待所有适配器写入完成
// 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
futures.forEach(future -> { RuntimeException exception = null;
for (Future<Boolean> future : futures) {
try { try {
if (!future.get()) { if (!future.get()) {
throw new RuntimeException("Outer adapter sync failed! "); exception = new RuntimeException("Outer adapter sync failed! ");
} }
} catch (Exception e) { } catch (Exception e) {
future.cancel(true); exception = new RuntimeException(e);
throw new RuntimeException(e);
} }
}); }
if (exception != null) {
throw exception;
}
}); });
} }