debezium/dbz#1553 Allow mining session lower bound to advance when time threshold reached

Signed-off-by: nathan-smit-1 <nathansm@pepkorit.com>

Author: nathan-smit-1

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/OracleConnectorConfig.java

@@ -511,6 +511,17 @@ public class OracleConnectorConfig extends HistorizedRelationalDatabaseConnector
             .withDescription(
                     "The maximum number of milliseconds that a LogMiner session lives for before being restarted. Defaults to 0 (indefinite until a log switch occurs)");
 
+    public static final Field LOG_MINING_WINDOW_MAX_MS = Field.create("log.mining.window.max.ms")
+            .withDisplayName("Maximum number of milliseconds that the mining window can span")
+            .withType(Type.LONG)
+            .withWidth(Width.SHORT)
+            .withImportance(Importance.LOW)
+            .withDefault(TimeUnit.MINUTES.toMillis(0))
+            .withValidation(Field::isNonNegativeInteger)
+            .withDescription("The maximum number of milliseconds that the mining window can span. " +
+                    "If a transaction remains open for longer than this duration, the mining window start SCN will be advanced " +
+                    "to minimize the window size, preventing it from growing indefinitely. Defaults to 0 (disabled).");
+
     public static final Field LOG_MINING_RESTART_CONNECTION = Field.create("log.mining.restart.connection")
             .withDisplayName("Restarts Oracle database connection when reaching maximum session time or database log switch")
             .withType(Type.BOOLEAN)
@@ -852,6 +863,7 @@ public class OracleConnectorConfig extends HistorizedRelationalDatabaseConnector
                     LOG_MINING_LOG_BACKOFF_INITIAL_DELAY_MS,
                     LOG_MINING_LOG_BACKOFF_MAX_DELAY_MS,
                     LOG_MINING_SESSION_MAX_MS,
+                    LOG_MINING_WINDOW_MAX_MS,
                     LOG_MINING_TRANSACTION_SNAPSHOT_BOUNDARY_MODE,
                     LOG_MINING_READ_ONLY,
                     LOG_MINING_FLUSH_TABLE_NAME,
@@ -944,6 +956,7 @@ public static ConfigDef configDef() {
     private final Duration logMiningInitialDelay;
     private final Duration logMiningMaxDelay;
     private final Duration logMiningMaximumSession;
+    private final Duration logMiningWindowMaxMs;
     private final TransactionSnapshotBoundaryMode logMiningTransactionSnapshotBoundaryMode;
     private final Boolean logMiningReadOnly;
     private final String logMiningFlushTableName;
@@ -1043,6 +1056,18 @@ public OracleConnectorConfig(Configuration config) {
         this.logMiningClientIdExcludes = Strings.setOfTrimmed(config.getString(LOG_MINING_CLIENTID_EXCLUDE_LIST), String::new);
         this.logMiningPathToDictionary = config.getString(LOG_MINING_PATH_DICTIONARY);
         this.logMiningUseCteQuery = config.getBoolean(LOG_MINING_USE_CTE_QUERY);
+
+        // Initialize logMiningWindowMaxMs, but disable if CTE is enabled as they are incompatible
+        final Duration configuredWindowMaxMs = Duration.ofMillis(config.getLong(LOG_MINING_WINDOW_MAX_MS));
+        if (this.logMiningUseCteQuery && !configuredWindowMaxMs.isZero()) {
+            LOGGER.warn("The log.mining.window.max.ms feature is not compatible with log.mining.use.cte.query. " +
+                    "The log.mining.window.max.ms feature will be disabled.");
+            this.logMiningWindowMaxMs = Duration.ZERO;
+        }
+        else {
+            this.logMiningWindowMaxMs = configuredWindowMaxMs;
+        }
+
         this.readonlyHostname = config.getString(LOG_MINING_READONLY_HOSTNAME);
         this.logMiningRedoThreadScnAdjustment = config.getInteger(LOG_MINING_REDO_THREAD_SCN_ADJUSTMENT);
         this.logMiningHashAreaSize = config.getLong(LOG_MINING_HASH_AREA_SIZE);
@@ -2001,6 +2026,13 @@ public Optional<Duration> getLogMiningMaximumSession() {
         return logMiningMaximumSession.toMillis() == 0L ? Optional.empty() : Optional.of(logMiningMaximumSession);
     }
 
+    /**
+     * @return the maximum duration for the mining window
+     */
+    public Duration getLogMiningWindowMaxMs() {
+        return logMiningWindowMaxMs;
+    }
+
     /**
      * @return how in-progress transactions are the snapshot boundary are to be handled.
      */

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/OracleOffsetContext.java

@@ -27,9 +27,12 @@ public class OracleOffsetContext extends CommonOffsetContext<SourceInfo> {
 
     public static final String SNAPSHOT_PENDING_TRANSACTIONS_KEY = "snapshot_pending_tx";
     public static final String SNAPSHOT_SCN_KEY = "snapshot_scn";
+    public static final String WINDOW_ADVANCE_ENABLED_KEY = "window_advance_enabled";
 
     private final Schema sourceInfoSchema;
 
+    private boolean windowAdvanceEnabled;
+
     private final TransactionContext transactionContext;
     private final IncrementalSnapshotContext<TableId> incrementalSnapshotContext;
 
@@ -52,12 +55,13 @@ private OracleOffsetContext(OracleConnectorConfig connectorConfig, Scn scn, Long
                                 Scn snapshotScn, Map<String, Scn> snapshotPendingTransactions, SnapshotType snapshot,
                                 boolean snapshotCompleted, TransactionContext transactionContext,
                                 IncrementalSnapshotContext<TableId> incrementalSnapshotContext,
-                                String transactionId, Long transactionSequence) {
+                                String transactionId, Long transactionSequence, boolean windowAdvanceEnabled) {
         super(new SourceInfo(connectorConfig), snapshotCompleted);
         sourceInfo.setScn(scn);
         sourceInfo.setScnIndex(scnIndex);
         sourceInfo.setTransactionId(transactionId);
         sourceInfo.setTransactionSequence(transactionSequence);
+        this.windowAdvanceEnabled = windowAdvanceEnabled;
         // It is safe to set this value to the supplied SCN, specifically for snapshots.
         // During streaming this value will be updated by the current event handler.
         sourceInfo.setEventScn(scn);
@@ -98,6 +102,7 @@ public static class Builder {
         private String transactionId;
         private Long transactionSequence;
         private CommitScn commitScn = CommitScn.empty();
+        private boolean windowAdvanceEnabled;
 
         public Builder logicalName(OracleConnectorConfig connectorConfig) {
             this.connectorConfig = connectorConfig;
@@ -164,10 +169,15 @@ public Builder commitScn(CommitScn commitScn) {
             return this;
         }
 
+        public Builder windowAdvanceEnabled(boolean windowAdvanceEnabled) {
+            this.windowAdvanceEnabled = windowAdvanceEnabled;
+            return this;
+        }
+
         public OracleOffsetContext build() {
             return new OracleOffsetContext(connectorConfig, scn, scnIndex, commitScn, lcrPosition, snapshotScn,
                     snapshotPendingTransactions, snapshot, snapshotCompleted, transactionContext,
-                    incrementalSnapshotContext, transactionId, transactionSequence);
+                    incrementalSnapshotContext, transactionId, transactionSequence, windowAdvanceEnabled);
         }
     }
 
@@ -218,6 +228,10 @@ public static Builder create() {
             }
         }
 
+        if (windowAdvanceEnabled) {
+            result.put(WINDOW_ADVANCE_ENABLED_KEY, true);
+        }
+
         return sourceInfo.isSnapshot() ? result : incrementalSnapshotContext.store(transactionContext.store(result));
     }
 
@@ -294,6 +308,18 @@ public void setSnapshotPendingTransactions(Map<String, Scn> snapshotPendingTrans
         this.snapshotPendingTransactions = snapshotPendingTransactions;
     }
 
+    public boolean isWindowAdvanceEnabled() {
+        return windowAdvanceEnabled;
+    }
+
+    /**
+     * Marks the window advance feature as having been enabled.
+     * Once set to true, this cannot be unset (until offsets are cleared).
+     */
+    public void setWindowAdvanceEnabled() {
+        this.windowAdvanceEnabled = true;
+    }
+
     public void setTransactionId(String transactionId) {
         sourceInfo.setTransactionId(transactionId);
     }
@@ -493,6 +519,17 @@ public static Long loadTransactionSequence(Map<String, ?> offset) {
         return readOffsetValue(offset, SourceInfo.TXSEQ_KEY, Long.class);
     }
 
+    /**
+     * Helper method to read whether window advance has been enabled from the offset map.
+     *
+     * @param offset the offset map
+     * @return true if window advance has ever been enabled, false otherwise
+     */
+    public static boolean loadWindowAdvanceEnabled(Map<String, ?> offset) {
+        Boolean value = readOffsetValue(offset, WINDOW_ADVANCE_ENABLED_KEY, Boolean.class);
+        return value != null && value;
+    }
+
     private static <T> T readOffsetValue(Map<String, ?> offsets, String key, Class<T> valueType) {
         final Object value = offsets.get(key);
         return valueType.isInstance(value) ? valueType.cast(value) : null;

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/buffered/BufferedLogMinerOracleOffsetContextLoader.java

@@ -40,6 +40,7 @@ public OracleOffsetContext load(Map<String, ?> offset) {
                 .transactionId(OracleOffsetContext.loadTransactionId(offset))
                 .transactionSequence(OracleOffsetContext.loadTransactionSequence(offset))
                 .incrementalSnapshotContext(SignalBasedIncrementalSnapshotContext.load(offset))
+                .windowAdvanceEnabled(OracleOffsetContext.loadWindowAdvanceEnabled(offset))
                 .build();
     }
 

debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/buffered/BufferedLogMinerStreamingChangeEventSource.java

@@ -74,6 +74,8 @@ public class BufferedLogMinerStreamingChangeEventSource extends AbstractLogMiner
 
     private static final Logger LOGGER = LoggerFactory.getLogger(BufferedLogMinerStreamingChangeEventSource.class);
     private static final Logger ABANDONED_DETAILS_LOGGER = LoggerFactory.getLogger(BufferedLogMinerStreamingChangeEventSource.class.getName() + ".AbandonedDetails");
+    private static final Logger WINDOW_ADVANCED = LoggerFactory.getLogger(BufferedLogMinerStreamingChangeEventSource.class.getName() + ".WindowAdvanced");
+
     private static final String NO_SEQUENCE_TRX_ID_SUFFIX = "ffffffff";
 
     private final String queryString;
@@ -82,6 +84,7 @@ public class BufferedLogMinerStreamingChangeEventSource extends AbstractLogMiner
 
     private Instant lastProcessedScnChangeTime = null;
     private Scn lastProcessedScn = Scn.NULL;
+    private Scn lastLoggedWindowAdvanceScn = Scn.NULL;
 
     public BufferedLogMinerStreamingChangeEventSource(OracleConnectorConfig connectorConfig,
                                                       OracleConnection jdbcConnection,
@@ -778,7 +781,13 @@ private ProcessResult calculateNewStartScn(Scn startScn, Scn endScn, Scn maxComm
 
         if (!minCacheScn.isNull()) {
             // Cache have values
-            final Scn miningSessionStartScn = minCacheScn.subtract(Scn.ONE);
+            // By default, the mining window starts at the oldest transaction in the cache.
+            // If window max is configured, it may be adjusted to not pin on long-running transactions.
+            final Scn miningSessionStartScn = applyWindowMaxAdjustment(
+                    minCacheScn.subtract(Scn.ONE),
+                    endScn,
+                    minCacheScn,
+                    minCacheScnChangeTime);
 
             getOffsetContext().setScn(miningSessionStartScn);
             getEventDispatcher().dispatchHeartbeatEvent(getPartition(), getOffsetContext());
@@ -801,6 +810,82 @@ private ProcessResult calculateNewStartScn(Scn startScn, Scn endScn, Scn maxComm
         }
     }
 
+    /**
+     * Adjusts the mining session start SCN based on the window max duration threshold.
+     * <p>
+     * When {@code log.mining.window.max.ms} is configured, this method prevents long-running
+     * transactions from pinning the mining window to an old position. If the oldest
+     * transaction in the cache exceeds the window threshold, this method finds the oldest
+     * transaction that falls within the acceptable window, or advances to the end SCN if all
+     * transactions are too old.
+     * <p>
+     * Long-running transactions will still be captured when they eventually commit, but they
+     * won't force the connector to re-mine an ever-growing window of redo logs.
+     *
+     * @param defaultStartScn the default start SCN (oldest transaction minus one)
+     * @param endScn the current end SCN of the mining window
+     * @param minCacheScn the SCN of the oldest transaction in the cache
+     * @param minCacheScnChangeTime the change time of the oldest transaction in the cache
+     * @return the adjusted start SCN, or {@code defaultStartScn} if no adjustment is needed
+     */
+    private Scn applyWindowMaxAdjustment(Scn defaultStartScn, Scn endScn,
+                                         Scn minCacheScn, Instant minCacheScnChangeTime) {
+        final Duration windowMaxMs = getConfig().getLogMiningWindowMaxMs();
+        if (windowMaxMs.toMillis() <= 0 || lastProcessedScnChangeTime == null) {
+            return defaultStartScn;
+        }
+
+        // Mark in offsets that the window advance feature has been enabled
+        if (!getOffsetContext().isWindowAdvanceEnabled()) {
+            getOffsetContext().setWindowAdvanceEnabled();
+        }
+
+        final Instant thresholdTime = lastProcessedScnChangeTime.minus(windowMaxMs);
+
+        // Check if the oldest transaction exceeds the window threshold
+        if (minCacheScnChangeTime == null || minCacheScnChangeTime.compareTo(thresholdTime) >= 0) {
+            // Oldest transaction is within the window, no adjustment needed
+            return defaultStartScn;
+        }
+
+        // The oldest transaction exceeds the threshold, find a suitable start SCN
+        // by looking for the oldest transaction that falls within the window
+        final Optional<LogMinerTransactionCache.ScnDetails> activeScnDetails = getTransactionCache()
+                .streamTransactionsAndReturn(stream -> stream
+                        .filter(t -> t.getChangeTime().compareTo(thresholdTime) >= 0)
+                        .map(t -> new LogMinerTransactionCache.ScnDetails(t.getStartScn(), t.getChangeTime()))
+                        .min(Comparator.comparing(LogMinerTransactionCache.ScnDetails::scn)));
+
+        Scn adjustedStartScn;
+        if (activeScnDetails.isPresent()) {
+            // Found a transaction within the window, use its start SCN
+            adjustedStartScn = activeScnDetails.get().scn().subtract(Scn.ONE);
+        }
+        else {
+            // All transactions are older than the window max duration
+            // Advance to the end SCN (like we do when the cache is empty)
+            adjustedStartScn = endScn.subtract(Scn.ONE);
+        }
+
+        // Safety check: never advance past the end SCN
+        final Scn maxAllowedScn = endScn.subtract(Scn.ONE);
+        if (adjustedStartScn.compareTo(maxAllowedScn) > 0) {
+            adjustedStartScn = maxAllowedScn;
+        }
+
+        // Log a warning, but only once per oldest transaction SCN to avoid flooding logs
+        if (!minCacheScn.equals(lastLoggedWindowAdvanceScn)) {
+            WINDOW_ADVANCED.warn("Mining window lower bound advanced past transaction at SCN {} to SCN {} " +
+                    "due to log.mining.window.max.ms threshold ({}ms). " +
+                    "Long-running transactions older than the threshold will continue to be captured " +
+                    "but won't pin the mining window.",
+                    minCacheScn, adjustedStartScn, windowMaxMs.toMillis());
+            lastLoggedWindowAdvanceScn = minCacheScn;
+        }
+
+        return adjustedStartScn;
+    }
+
     /**
      * Calculates the smallest system change number currently in the transaction cache, if any exist.
      *

debezium-connector-oracle/src/test/java/io/debezium/connector/oracle/logminer/buffered/AbstractBufferedLogMinerStreamingChangeEventSourceIT.java

@@ -7,6 +7,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -232,4 +233,58 @@ public void shouldLogAdditionalDetailsForAbandonedTransaction() throws Exception
             TestHelper.dropTable(connection, "dbz8044");
         }
     }
+
+    @Test
+    @FixFor("DBZ-1553")
+    public void shouldAdvanceMiningWindowForLongRunningTransaction() throws Exception {
+        TestHelper.dropTable(connection, "dbz1553");
+        try {
+            connection.execute("CREATE TABLE dbz1553 (id numeric(9,0) primary key, data varchar2(50))");
+            TestHelper.streamTable(connection, "dbz1553");
+
+            // Configure the connector with a 30 second window max
+            Configuration config = getBufferImplementationConfig()
+                    .with(OracleConnectorConfig.TABLE_INCLUDE_LIST, "DEBEZIUM\\.DBZ1553")
+                    .with(OracleConnectorConfig.LOG_MINING_WINDOW_MAX_MS, "30000")
+                    .with(OracleConnectorConfig.SNAPSHOT_MODE, OracleConnectorConfig.SnapshotMode.NO_DATA)
+                    .build();
+
+            final LogInterceptor logInterceptor = new LogInterceptor(BufferedLogMinerStreamingChangeEventSource.class);
+            start(OracleConnector.class, config);
+            assertConnectorIsRunning();
+
+            waitForStreamingRunning(TestHelper.CONNECTOR_NAME, TestHelper.SERVER_NAME);
+
+            // Start a long-running transaction that will not be committed
+            connection.executeWithoutCommitting("INSERT INTO dbz1553 (id,data) values (1, 'long-running')");
+
+            // Wait for the window threshold to be exceeded and the mining window to be advanced.
+            // The log message should appear once the mining window lower bound is moved past the
+            // long-running transaction.
+            Awaitility.await()
+                    .atMost(Duration.ofMinutes(2))
+                    .pollInterval(Duration.ofSeconds(5))
+                    .until(() -> logInterceptor.containsWarnMessage("Mining window lower bound advanced"));
+
+            // Verify the warning message indicates the window was advanced due to the threshold
+            assertThat(logInterceptor.containsWarnMessage("due to log.mining.window.max.ms threshold")).isTrue();
+
+            // Now commit the long-running transaction
+            connection.commit();
+
+            // Consume the record to verify the transaction was fully captured
+            SourceRecords records = consumeRecordsByTopic(1);
+            assertThat(records.allRecordsInOrder()).hasSize(1);
+
+            List<SourceRecord> tableRecords = records.recordsForTopic("server1.DEBEZIUM.DBZ1553");
+            assertThat(tableRecords).hasSize(1);
+
+            Struct after = ((Struct) tableRecords.get(0).value()).getStruct(Envelope.FieldName.AFTER);
+            assertThat(after.get("ID")).isEqualTo(1);
+            assertThat(after.get("DATA")).isEqualTo("long-running");
+        }
+        finally {
+            TestHelper.dropTable(connection, "dbz1553");
+        }
+    }
 }