[FLINK-34660][checkpoint] Avoid randomizing changelog-related config …

…when it's pre-defined (#24488)
apache · Mar 14, 2024 · aa71589 · aa71589
1 parent 0da60ca
commit aa71589
Showing 1 changed file with 12 additions and 8 deletions.
diff --git a/...flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java b/...flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
@@ -124,15 +124,17 @@ private static void randomizeConfiguration(MiniCluster miniCluster, Configuratio
         }
 
         // randomize ITTests for enabling state change log
-        if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(STATE_CHANGE_LOG_CONFIG_ON)) {
-            if (!conf.contains(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG)) {
+        if (!conf.contains(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG)) {
+            if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(STATE_CHANGE_LOG_CONFIG_ON)) {
                 conf.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, true);
-                miniCluster.overrideRestoreModeForChangelogStateBackend();
+            } else if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(STATE_CHANGE_LOG_CONFIG_RAND)) {
+                randomize(conf, StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, true, false);
             }
-        } else if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(STATE_CHANGE_LOG_CONFIG_RAND)) {
-            boolean enabled =
-                    randomize(conf, StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, true, false);
-            if (enabled) {
+        }
+
+        // randomize periodic materialization when enabling state change log
+        if (conf.get(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG)) {
+            if (!conf.contains(StateChangelogOptions.PERIODIC_MATERIALIZATION_ENABLED)) {
                 // More situations about enabling periodic materialization should be tested
                 randomize(
                         conf,
@@ -141,15 +143,17 @@ private static void randomizeConfiguration(MiniCluster miniCluster, Configuratio
                         true,
                         true,
                         false);
+            }
+            if (!conf.contains(StateChangelogOptions.PERIODIC_MATERIALIZATION_INTERVAL)) {
                 randomize(
                         conf,
                         StateChangelogOptions.PERIODIC_MATERIALIZATION_INTERVAL,
                         Duration.ofMillis(100),
                         Duration.ofMillis(500),
                         Duration.ofSeconds(1),
                         Duration.ofSeconds(5));
-                miniCluster.overrideRestoreModeForChangelogStateBackend();
             }
+            miniCluster.overrideRestoreModeForChangelogStateBackend();
         }
     }