DBZ-7445 LogMiner batch size does not increase automatically

The condition for increasing the LogMiner batch size must depend on log.mining.batch.size.max, and not on log.mining.batch.size.min.
couchbase · Feb 5, 2024 · dc18a48 · dc18a48
1 parent 3d1a1e8
commit dc18a48
Showing 1 changed file with 8 additions and 6 deletions.
diff --git a/...c/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java b/...c/main/java/io/debezium/connector/oracle/logminer/LogMinerStreamingChangeEventSource.java
@@ -680,19 +680,21 @@ public void endMiningSession(OracleConnection connection, OracleOffsetContext of
     }
 
     private void updateBatchSize(boolean increment) {
-        if (increment && currentBatchSize < connectorConfig.getLogMiningBatchSizeMin()) {
-            currentBatchSize += connectorConfig.getLogMiningBatchSizeMin();
-            if (currentBatchSize == connectorConfig.getLogMiningBatchSizeMax()) {
+        int batchSizeMin = connectorConfig.getLogMiningBatchSizeMin();
+        int batchSizeMax = connectorConfig.getLogMiningBatchSizeMax();
+        if (increment && currentBatchSize < batchSizeMax) {
+            currentBatchSize = Math.min(currentBatchSize + batchSizeMin, batchSizeMax);
+            if (currentBatchSize == batchSizeMax) {
                 LOGGER.info("The connector is now using the maximum batch size {} when querying the LogMiner view.{}",
                         currentBatchSize,
                         connectorConfig.isLobEnabled() ? "" : " This could be indicate of a large SCN gap.");
             }
         }
-        else if (!increment && currentBatchSize > connectorConfig.getLogMiningBatchSizeMin()) {
-            currentBatchSize -= connectorConfig.getLogMiningBatchSizeMin();
+        else if (!increment && currentBatchSize > batchSizeMin) {
+            currentBatchSize = Math.max(currentBatchSize - batchSizeMin, batchSizeMin);
         }
 
-        if (currentBatchSize != connectorConfig.getLogMiningBatchSizeMax()) {
+        if (currentBatchSize != batchSizeMax) {
             LOGGER.debug("Updated batch size window, using batch size {}", currentBatchSize);
         }