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

fixed issue #782 , 重复了exception

Browse Source
This commit is contained in:
七锋
2018-07-26 16:05:33 +08:00
parent d624358cb5
commit 8cfb70e080
1 changed files with 4 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -4,7 +4,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.LockSupport;
import com.alibaba.otter.canal.common.CanalException;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
@@ -64,7 +63,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private ExecutorService stageExecutor; private ExecutorService stageExecutor;
private String destination; private String destination;
private volatile CanalParseException exception; private volatile CanalParseException exception;
private volatile RuntimeException incident;
public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
EventTransactionBuffer transactionBuffer, String destination){ EventTransactionBuffer transactionBuffer, String destination){
@@ -79,7 +77,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
public void start() { public void start() {
super.start(); super.start();
this.exception = null; this.exception = null;
this.incident = null;
this.disruptorMsgBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(), this.disruptorMsgBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(),
ringBufferSize, ringBufferSize,
new BlockingWaitStrategy()); new BlockingWaitStrategy());
@@ -156,13 +153,13 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
return false; return false;
} }
/* 由于改为processor仅终止自身stage而不是stop，那么需要由incident标识coprocessor是否正常工作。 /**
* 由于改为processor仅终止自身stage而不是stop，那么需要由incident标识coprocessor是否正常工作。
* 让dump线程能够及时感知 * 让dump线程能够及时感知
*/ */
if (incident != null) { if (exception != null) {
throw incident; throw exception;
} }
boolean interupted = false; boolean interupted = false;
do { do {
try { try {
@@ -180,11 +177,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
interupted = Thread.interrupted(); interupted = Thread.interrupted();
} }
} while (!interupted && isStart()); } while (!interupted && isStart());
if (exception != null) {
throw exception;
}
return isStart(); return isStart();
} }
@@ -405,12 +397,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
@Override @Override
public void handleEventException(final Throwable ex, final long sequence, final Object event) { public void handleEventException(final Throwable ex, final long sequence, final Object event) {
/*
* 若异常不是onEvent body里产生的。如果仅仅抛出去，onShutdown会调用stop()，与dump线程的reset()产生竞争，导致向terminated ExecutorService提交新task
* #771场景中的异常是InterruptedException，由shutdownNow产生。
*/
incident = new RuntimeException(ex);
throw incident;
} }
@Override @Override