Merge branch 'master' of http://git.navis.lan/scm/n4fra/debezium

AndreyIg · Mar 11, 2020 · 76f9f80 · 76f9f80
2 parents 77e567e + af6f8a3
commit 76f9f80
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 2 deletions.
diff --git a/...cle/src/main/java/io/debezium/connector/oracle/logminer/LogMinerQueryResultProcessor.java b/...cle/src/main/java/io/debezium/connector/oracle/logminer/LogMinerQueryResultProcessor.java
@@ -128,7 +128,7 @@ public int processResult(ResultSet resultSet) {
 
             // DDL
             if (operationCode == RowMapper.DDL) {
-                LOGGER.debug("DDL,  {}", logMessage);
+                LOGGER.debug("DDL: {}, REDO_SQL {}", logMessage, redo_sql);
                 continue;
                 // todo parse, add to the collection.
             }

diff --git a/...c/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java b/...c/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java
@@ -153,8 +153,9 @@ public void execute(ChangeEventSourceContext context) throws InterruptedExceptio
                             Optional<Long> oldestScnToAbandonTransactions = LogMinerHelper.getLastScnFromTheOldestOnlineRedo(connection, offsetContext.getScn());
                             oldestScnToAbandonTransactions.ifPresent(nextOldestScn -> {
                                 transactionalBuffer.abandonLongTransactions(nextOldestScn);
-                                offsetContext.setScn(nextOldestScn);
                                 LOGGER.debug("After abandoning, offset before: {}, offset after:{}", offsetContext.getScn(), nextOldestScn);
+                                offsetContext.setScn(nextOldestScn);
+                                lastProcessedScn = transactionalBuffer.getLargestScn().equals(BigDecimal.ZERO) ? nextScn : transactionalBuffer.getLargestScn().longValue();
                             });
 
                             LogMinerHelper.setRedoLogFilesForMining(connection, lastProcessedScn);

diff --git a/...ector-oracle/src/main/java/io/debezium/connector/oracle/logminer/TransactionalBuffer.java b/...ector-oracle/src/main/java/io/debezium/connector/oracle/logminer/TransactionalBuffer.java
@@ -82,6 +82,7 @@ public final class TransactionalBuffer {
      * @return largest last SCN in the buffer among all transactions
      */
     public BigDecimal getLargestScn() {
+        calculateLargestScn();
         return largestScn;
     }