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

DBZ-3676 When LOB support is disabled, use adaptive SCN mining windows

Browse Source
This commit is contained in:
Chris Cranford 2021-06-25 11:47:30 -04:00 committed by Gunnar Morling
parent 09fa730d98
commit 01e87c11ed
2 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java
View File

@ -185,7 +185,16 @@ public void execute(ChangeEventSourceContext context, OracleOffsetContext offset
Duration lastDurationOfBatchCapturing = stopwatch.stop().durations().statistics().getTotal();
streamingMetrics.setLastDurationOfBatchCapturing(lastDurationOfBatchCapturing);
processor.processResult(rs);
startScn = transactionalBuffer.updateOffsetContext(offsetContext, dispatcher);
if (connectorConfig.isLobEnabled()) {
startScn = transactionalBuffer.updateOffsetContext(offsetContext, dispatcher);
}
else {
if (transactionalBuffer.isEmpty()) {
LOGGER.debug("Buffer is empty, updating offset SCN to {}", endScn);
offsetContext.setScn(endScn);
}
startScn = endScn;
}
}
streamingMetrics.setCurrentBatchProcessingTime(Duration.between(start, Instant.now()));

6
debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/TransactionalBuffer.java
View File

@ -319,8 +319,10 @@ boolean commit(String transactionId, Scn scn, OracleOffsetContext offsetContext,
maxCommittedScn = lastCommittedScn;
}
// cache recent transaction and commit scn for handling offset updates
recentlyCommittedTransactionIds.add(new RecentlyCommittedTransaction(transaction, scn));
if (connectorConfig.isLobEnabled()) {
// cache recent transaction and commit scn for handling offset updates
recentlyCommittedTransactionIds.add(new RecentlyCommittedTransaction(transaction, scn));
}
}
catch (InterruptedException e) {
LogMinerHelper.logError(streamingMetrics, "Commit interrupted", e);