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

DBZ-6275 Correctly sync Infinispan cache on event counter reset

Browse Source
This commit is contained in:
Chris Cranford 2023-04-01 15:34:11 -04:00 committed by Chris Cranford
parent 6419c0e704
commit 3bb82d6b73
2 changed files with 13 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/processor/AbstractLogMinerEventProcessor.java
View File

@ -343,17 +343,21 @@ protected void handleMissingScn(LogMinerEventRow row) {
*/
protected void handleStart(LogMinerEventRow row) {
final String transactionId = row.getTransactionId();
final AbstractTransaction transaction = getTransactionCache().get(transactionId);
final T transaction = getTransactionCache().get(transactionId);
if (transaction == null && !isRecentlyProcessed(transactionId)) {
getTransactionCache().put(transactionId, createTransaction(row));
metrics.setActiveTransactions(getTransactionCache().size());
}
else if (transaction != null && !isRecentlyProcessed(transactionId)) {
LOGGER.trace("Transaction {} is not yet committed and START event detected.", transactionId);
transaction.start();
resetTransactionToStart(transaction);
}
}
protected void resetTransactionToStart(T transaction) {
transaction.start();
}
/**
* Handle processing a LogMinerEventRow for a {@code COMMIT} event.
*

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

@ -234,6 +234,13 @@ protected void finalizeTransactionRollback(String transactionId, Scn rollbackScn
getProcessedTransactionsCache().put(transactionId, rollbackScn.toString());
}
@Override
protected void resetTransactionToStart(InfinispanTransaction transaction) {
super.resetTransactionToStart(transaction);
// Flush the change created by the super class to the transaction cache
getTransactionCache().put(transaction.getTransactionId(), transaction);
}
@Override
protected void handleSchemaChange(LogMinerEventRow row) throws InterruptedException {
super.handleSchemaChange(row);