[hotfix] Adjust the writeBatch estimate logic to avoid buffer size always tending to decrease (#1076)

Author: swuferhong

fluss-client/src/main/java/com/alibaba/fluss/client/write/DynamicWriteBatchSizeEstimator.java

@@ -34,7 +34,7 @@ public class DynamicWriteBatchSizeEstimator {
 
     private static final Logger LOG = LoggerFactory.getLogger(DynamicWriteBatchSizeEstimator.class);
 
-    private static final double RATIO_TO_INCREASE_BATCH_SIZE = 0.9d;
+    private static final double RATIO_TO_INCREASE_BATCH_SIZE = 0.8d;
     private static final double RATIO_TO_DECREASE_BATCH_SIZE = 0.5d;
     private final int maxBatchSize;
     private final int pageSize;
@@ -63,18 +63,14 @@ public void updateEstimation(PhysicalTablePath physicalTablePath, int observedBa
 
         int estimatedBatchSize =
                 estimatedBatchSizeMap.getOrDefault(physicalTablePath, maxBatchSize);
-        int newEstimatedBatchSize;
+        int newEstimatedBatchSize = estimatedBatchSize;
         if (observedBatchSize >= estimatedBatchSize
                 || observedBatchSize > estimatedBatchSize * RATIO_TO_INCREASE_BATCH_SIZE) {
-            // To increase 1%
+            // To increase 10%
             newEstimatedBatchSize = Math.min((int) (estimatedBatchSize * 1.1), maxBatchSize);
-        } else if (observedBatchSize < estimatedBatchSize * RATIO_TO_INCREASE_BATCH_SIZE
-                && observedBatchSize > estimatedBatchSize * RATIO_TO_DECREASE_BATCH_SIZE) {
+        } else if (observedBatchSize < estimatedBatchSize * RATIO_TO_DECREASE_BATCH_SIZE) {
             // To decrease 5%
-            newEstimatedBatchSize = Math.max((int) (estimatedBatchSize * 0.95), pageSize);
-        } else {
-            // To decrease 10%
-            newEstimatedBatchSize = Math.max((int) (estimatedBatchSize * 0.9), pageSize);
+            newEstimatedBatchSize = Math.max((int) (estimatedBatchSize * 0.95), 2 * pageSize);
         }
 
         estimatedBatchSizeMap.put(physicalTablePath, newEstimatedBatchSize);

fluss-client/src/test/java/com/alibaba/fluss/client/write/DynamicWriteBatchSizeEstimatorTest.java

@@ -39,32 +39,36 @@ void testEstimator() {
         assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(1000);
 
         estimator = new DynamicWriteBatchSizeEstimator(true, 1000, 100);
-        // test decrease 10%
+        // test decrease 5%
         estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 450);
-        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(900);
+        int expectedSize = 950;
+        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH))
+                .isEqualTo(expectedSize);
 
-        // test decrease 5%
-        estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, (int) (900 * 0.9) - 10);
-        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(855);
+        // test increase 5%
+        estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 350);
+        expectedSize = (int) (950 * 0.95);
+        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH))
+                .isEqualTo(expectedSize);
 
-        // test increase 1%
-        estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 852);
+        // test increase 10%
+        estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 930);
         assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH))
-                .isEqualTo((int) (855 * 1.1));
+                .isEqualTo((int) (expectedSize * 1.1));
     }
 
     @Test
     void testMinDecreaseToPageSize() {
         int estimatedSize = estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH);
         estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 1000);
-        while (estimatedSize > 100) {
+        while (estimatedSize > 2 * 100) {
             estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, (int) (estimatedSize * 0.5) - 10);
             estimatedSize = estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH);
         }
 
-        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(100);
+        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(200);
         estimator.updateEstimation(DATA1_PHYSICAL_TABLE_PATH, 0);
-        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(100);
+        assertThat(estimator.getEstimatedBatchSize(DATA1_PHYSICAL_TABLE_PATH)).isEqualTo(200);
     }
 
     @Test