private void abandonOldTransactionsIfExist(
            OracleConnection connection,
            OracleOffsetContext offsetContext,
            TransactionalBuffer transactionalBuffer) {
        Duration transactionRetention = connectorConfig.getLogMiningTransactionRetention();
        if (!Duration.ZERO.equals(transactionRetention)) {
            final Scn offsetScn = offsetContext.getScn();
            Optional<Scn> lastScnToAbandonTransactions =
                    getLastScnToAbandon(connection, offsetScn, transactionRetention);
            lastScnToAbandonTransactions.ifPresent(
                    thresholdScn -> {
                        transactionalBuffer.abandonLongTransactions(thresholdScn, offsetContext);
                        offsetContext.setScn(thresholdScn);
                        startScn = endScn;
                    });
        }
    }
This is the loop to get changes from LogMiner.

@param context change event source context
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java