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

adapter 分组间线程统一抛异常

Browse Source
This commit is contained in:
mcy
2018-12-28 17:30:04 +08:00
parent 150a13824b
commit 7ea1468a8d
1 changed files with 20 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -73,16 +73,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;
}
}); });
} }
@@ -113,22 +116,25 @@ 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;
}
}); });
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage, CanalMQConnector connector, protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage,
ExecutorService workerExecutor) { CanalMQConnector connector, ExecutorService workerExecutor) {
try { try {
List<?> messages; List<?> messages;
if (!flatMessage) { if (!flatMessage) {
@@ -157,7 +163,7 @@ public abstract class AbstractCanalAdapterWorker {
} }
connector.ack(); connector.ack();
} }
return true; return true;
} catch (Throwable e) { } catch (Throwable e) {
if (i == retry - 1) { if (i == retry - 1) {
connector.ack(); connector.ack();
@@ -173,7 +179,7 @@ public abstract class AbstractCanalAdapterWorker {
// ignore // ignore
} }
} }
return false; return false;
} }
/** /**