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

DBZ-7169 - populating processed transaction cache only if lobs enabled - in sync with memory event processor

Browse Source
This commit is contained in:
Jiri Kulhanek 2023-11-23 20:15:40 +01:00 committed by Chris Cranford
parent f76a78ee2f
commit 71d80af0b7
1 changed files with 8 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/processor/infinispan/AbstractInfinispanLogMinerEventProcessor.java
View File

@ -254,7 +254,9 @@ public LogMinerEvent next() {
@Override @Override
protected void finalizeTransactionCommit(String transactionId, Scn commitScn) { protected void finalizeTransactionCommit(String transactionId, Scn commitScn) {
// cache recently committed transactions by transaction id // cache recently committed transactions by transaction id
getProcessedTransactionsCache().put(transactionId, commitScn.toString()); if (getConfig().isLobEnabled()) {
getProcessedTransactionsCache().put(transactionId, commitScn.toString());
}
} }
@Override @Override
@ -264,7 +266,9 @@ protected void finalizeTransactionRollback(String transactionId, Scn rollbackScn
removeEventsWithTransaction(transaction); removeEventsWithTransaction(transaction);
getTransactionCache().remove(transactionId); getTransactionCache().remove(transactionId);
} }
getProcessedTransactionsCache().put(transactionId, rollbackScn.toString()); if (getConfig().isLobEnabled()) {
getProcessedTransactionsCache().put(transactionId, rollbackScn.toString());
}
} }
@Override @Override
@ -346,8 +350,7 @@ protected Scn calculateNewStartScn(Scn endScn, Scn maxCommittedScn) throws Inter
purgeCache(minCacheScn); purgeCache(minCacheScn);
} }
else { else {
getProcessedTransactionsCache().clear(); getSchemaChangesCache().entrySet().removeIf(e -> true);
getSchemaChangesCache().clear();
} }
if (getConfig().isLobEnabled()) { if (getConfig().isLobEnabled()) {
@ -357,6 +360,7 @@ protected Scn calculateNewStartScn(Scn endScn, Scn maxCommittedScn) throws Inter
} }
else { else {
if (!minCacheScn.isNull()) { if (!minCacheScn.isNull()) {
getProcessedTransactionsCache().entrySet().removeIf(entry -> Scn.valueOf(entry.getValue()).compareTo(minCacheScn) < 0);
offsetContext.setScn(minCacheScn.subtract(Scn.valueOf(1))); offsetContext.setScn(minCacheScn.subtract(Scn.valueOf(1)));
dispatcher.dispatchHeartbeatEvent(partition, offsetContext); dispatcher.dispatchHeartbeatEvent(partition, offsetContext);
} }