[FLINK-15146][core][ttl] Fix check that incremental cleanup size must…

… be greater than zero.

At the moment, we allow zero value for the TTL incremental cleanup size.
Although technically zero value will not break the cleanup but practically there will be no cleanup.
We change cleanupSize to be strictly greater than zero to avoid confusion.

flink-core/src/main/java/org/apache/flink/api/common/state/StateTtlConfig.java

@@ -413,8 +413,8 @@ public static class IncrementalCleanupStrategy implements CleanupStrategies.Clea
 
  private IncrementalCleanupStrategy(int cleanupSize, boolean runCleanupForEveryRecord) {
  Preconditions.checkArgument(
- cleanupSize >= 0,
- "Number of incrementally cleaned up state entries cannot be negative.");
+ cleanupSize > 0,
+ "Number of incrementally cleaned up state entries should be positive.");
  this.cleanupSize = cleanupSize;
  this.runCleanupForEveryRecord = runCleanupForEveryRecord;
  }

flink-core/src/test/java/org/apache/flink/api/common/state/StateTtlConfigTest.java

@@ -22,8 +22,12 @@
 import org.apache.flink.api.common.state.StateTtlConfig.RocksdbCompactFilterCleanupStrategy;
 import org.apache.flink.api.common.time.Time;
 
+import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.Arrays;
+import java.util.List;
+
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
@@ -71,4 +75,20 @@ public void testStateTtlConfigBuildWithCleanupInBackground() {
  assertThat(incrementalCleanupStrategy.runCleanupForEveryRecord(), is(false));
  assertThat(rocksdbCleanupStrategy.getQueryTimeAfterNumEntries(), is(1000L));
  }
+
+ @Test
+ public void testStateTtlConfigBuildWithNonPositiveCleanupIncrementalSize() {
+ List<Integer> illegalCleanUpSizes = Arrays.asList(0, -2);
+
+ for (Integer illegalCleanUpSize : illegalCleanUpSizes) {
+ try {
+ StateTtlConfig ttlConfig =
+ StateTtlConfig.newBuilder(Time.seconds(1))
+ .cleanupIncrementally(illegalCleanUpSize, false)
+ .build();
+ Assert.fail();
+ } catch (IllegalArgumentException e) {
+ }
+ }
+ }
 }