[FLINK-23279][tests] Randomly use Changelog Backend in tests

Unless LocalRecovery is enabled which is currently
not supported by the Changelog State Backend.

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

@@ -658,6 +658,11 @@ protected boolean useLocalCommunication() {
  return miniClusterConfiguration.getNumTaskManagers() == 1;
  }
 
+ @VisibleForTesting
+ public Configuration getConfiguration() {
+ return miniClusterConfiguration.getConfiguration();
+ }
+
  @GuardedBy("lock")
  private Collection<? extends CompletableFuture<Void>> terminateTaskManagers() {
  final Collection<CompletableFuture<Void>> terminationFutures =

flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java

@@ -19,6 +19,7 @@
 package org.apache.flink.streaming.util;
 
 import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -32,6 +33,7 @@
 import java.util.Collection;
 import java.util.Collections;
 
+import static org.apache.flink.configuration.CheckpointingOptions.LOCAL_RECOVERY;
 import static org.apache.flink.runtime.testutils.PseudoRandomValueSelector.randomize;
 
 /** A {@link StreamExecutionEnvironment} that executes its jobs on {@link MiniCluster}. */
@@ -95,10 +97,18 @@ public static void setAsContext(
  Duration.ofSeconds(2));
  }
  if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(STATE_CHANGE_LOG_CONFIG_ON)) {
- conf.set(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG, true);
+ if (isConfigurationSupportedByChangelog(miniCluster.getConfiguration())) {
+ conf.set(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG, true);
+ }
  } else if (STATE_CHANGE_LOG_CONFIG.equalsIgnoreCase(
  STATE_CHANGE_LOG_CONFIG_RAND)) {
- randomize(conf, CheckpointingOptions.ENABLE_STATE_CHANGE_LOG, true, false);
+ if (isConfigurationSupportedByChangelog(miniCluster.getConfiguration())) {
+ randomize(
+ conf,
+ CheckpointingOptions.ENABLE_STATE_CHANGE_LOG,
+ true,
+ false);
+ }
  }
  if (RANDOMIZE_BUFFER_DEBLOAT_CONFIG) {
  randomize(conf, TaskManagerOptions.BUFFER_DEBLOAT_ENABLED, true, false);
@@ -110,6 +120,10 @@ public static void setAsContext(
  initializeContextEnvironment(factory);
  }
 
+ private static boolean isConfigurationSupportedByChangelog(Configuration configuration) {
+ return !configuration.get(LOCAL_RECOVERY);
+ }
+
  /**
  * Sets the streaming context environment to a TestStreamEnvironment that runs its programs on
  * the given cluster with the given default parallelism.

pom.xml

@@ -1531,7 +1531,7 @@ under the License.
  <checkpointing.randomization>true</checkpointing.randomization>
  <buffer-debloat.randomization>true</buffer-debloat.randomization>
  <!-- on, unset, or random -->
- <checkpointing.changelog>unset</checkpointing.changelog>
+ <checkpointing.changelog>random</checkpointing.changelog>
  <project.basedir>${project.basedir}</project.basedir>
  <!--suppress MavenModelInspection -->
  <test.randomization.seed>${test.randomization.seed}</test.randomization.seed>