debezium/dbz#1279 Fix archive log only mode logic

Signed-off-by: Chris Cranford <chris@hibernate.org>

Author: Naros

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/AbstractLogMinerStreamingChangeEventSource.java

@@ -835,7 +835,7 @@ protected boolean waitForRangeAvailabilityInArchiveLogs(Scn startScn, Scn endScn
             }
         }
         else if (isNoDataProcessedInBatchAndAtEndOfArchiveLogs()) {
-            if (endScn.compareTo(getMaximumArchiveLogsScn()) == 0) {
+            if (endScn.compareTo(getMaximumArchiveLogsScn(startScn)) == 0) {
                 // Prior iteration mined up to the last entry in the archive logs and no data was returned.
                 return isArchiveLogOnlyModeAndScnIsNotAvailable(endScn.add(Scn.ONE));
             }
@@ -861,7 +861,7 @@ else if (isNoDataProcessedInBatchAndAtEndOfArchiveLogs()) {
      * @throws SQLException if a database exception is thrown
      */
     protected Scn calculateUpperBounds(Scn lowerBoundsScn, Scn previousUpperBounds, Scn currentScn) throws SQLException {
-        final Scn maximumScn = getConfig().isArchiveLogOnlyMode() ? getMaximumArchiveLogsScn() : currentScn;
+        final Scn maximumScn = getConfig().isArchiveLogOnlyMode() ? getMaximumArchiveLogsScn(lowerBoundsScn) : currentScn;
 
         final Scn maximumBatchScn = lowerBoundsScn.add(Scn.valueOf(metrics.getBatchSize()));
         final Scn defaultBatchSizeScn = Scn.valueOf(connectorConfig.getLogMiningBatchSizeDefault());
@@ -1072,11 +1072,9 @@ protected Scn getCurrentScn() throws SQLException {
      *
      * @return the maximum SCN, never {@code null}
      */
-    protected Scn getMaximumArchiveLogsScn() {
-        final List<LogFile> archiveLogs = (currentLogFiles == null)
-                ? Collections.emptyList()
-                : currentLogFiles.stream().filter(LogFile::isArchive).toList();
-
+    protected Scn getMaximumArchiveLogsScn(Scn startScn) throws SQLException {
+        // It is safe to query these in real-time
+        final List<LogFile> archiveLogs = logCollector.getLogs(startScn).stream().filter(LogFile::isArchive).toList();
         if (archiveLogs.isEmpty()) {
             throw new DebeziumException("Cannot get maximum archive log SCN as no archive logs are present.");
         }
@@ -1167,11 +1165,6 @@ protected void collectLogs(Scn lowerBoundsScn, Scn upperBoundsScn) throws SQLExc
             }
 
             sessionLogFiles = sessionLogFilesToAdd;
-
-            currentRedoLogSequences = currentLogFiles.stream()
-                    .filter(LogFile::isCurrent)
-                    .map(LogFile::getSequence)
-                    .toList();
         }
 
         metrics.setRedoLogStatuses(jdbcConnection.queryAndMap(
@@ -1205,6 +1198,12 @@ protected void applyLogsToSession(boolean postMiningSessionEnded) throws SQLExce
                 sessionContext.addLogFile(logFile.getFileName());
             }
 
+            // These need to be updated when we prepare the session so that log switch check works
+            currentRedoLogSequences = currentLogFiles.stream()
+                    .filter(LogFile::isCurrent)
+                    .map(LogFile::getSequence)
+                    .toList();
+
             metrics.setMinedLogFileNames(sessionLogFiles.stream()
                     .map(LogFile::getFileName)
                     .collect(Collectors.toSet()));

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/unbuffered/UnbufferedLogMinerStreamingChangeEventSource.java

@@ -116,21 +116,21 @@ protected void executeLogMiningStreaming() throws Exception {
             // except once per iteration.
             databaseOffset = getMetrics().getDatabaseOffset();
 
-            minLogScn = computeResumeScnAndUpdateOffsets(minLogScn, minCommitScn);
-            getMetrics().setOffsetScn(minLogScn);
-
             Scn currentScn = getCurrentScn();
             getMetrics().setCurrentScn(currentScn);
 
+            collectLogs(minLogScn, getCurrentScn());
+
+            minLogScn = computeResumeScnAndUpdateOffsets(minLogScn, minCommitScn);
+            getMetrics().setOffsetScn(minLogScn);
+
             upperBoundsScn = calculateUpperBounds(minLogScn, upperBoundsScn, currentScn);
             if (upperBoundsScn.isNull()) {
                 LOGGER.debug("Delaying mining transaction logs by one iteration");
                 pauseBetweenMiningSessions();
                 continue;
             }
 
-            collectLogs(minLogScn, getCurrentScn());
-
             if (firstBatch) {
                 applyLogsToSession(false);
                 firstBatch = false;