alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-2124 Do not call markBatchFinished on failure

Browse Source
This commit is contained in:
Chris Cranford 2020-06-01 09:54:22 -04:00 committed by Gunnar Morling
parent 966b5ce7f0
commit 6b5ac2494b
1 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
debezium-embedded/src/main/java/io/debezium/embedded/EmbeddedEngine.java
View File

@ -459,23 +459,19 @@ private static ChangeConsumer buildDefaultChangeConsumer(Consumer<SourceRecord>
*/
@Override
public void handleBatch(List<SourceRecord> records, DebeziumEngine.RecordCommitter<SourceRecord> committer) throws InterruptedException {
try {
for (SourceRecord record : records) {
try {
consumer.accept(record);
committer.markProcessed(record);
}
catch (StopConnectorException | StopEngineException ex) {
// ensure that we mark the record as finished
// in this case
committer.markProcessed(record);
throw ex;
}
for (SourceRecord record : records) {
try {
consumer.accept(record);
committer.markProcessed(record);
}
catch (StopConnectorException | StopEngineException ex) {
// ensure that we mark the record as finished
// in this case
committer.markProcessed(record);
throw ex;
}
}
finally {
committer.markBatchFinished();
}
committer.markBatchFinished();
}
};
}