[FLINK-17258][network][test] Run ClassLoaderITCase#testDisposeSavepoi…

…ntWithCustomKvState without unaligned checkpoints This test needs many concurent checkpoint & savepoints and this is currently not supported with unaligned checkpoints.
spotify · May 21, 2020 · b3a7851 · b3a7851
1 parent 840a4d9
commit b3a7851
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
@@ -313,7 +313,9 @@ public void testDisposeSavepointWithCustomKvState() throws Exception {
  String.valueOf(parallelism),
  checkpointDir.toURI().toString(),
  "5000",
- outputDir.toURI().toString()})
+ outputDir.toURI().toString(),
+ "false" // Disable unaligned checkpoints as this test is triggering concurrent savepoints/checkpoints
+ })
  .build();
 
  TestStreamEnvironment.setAsContext(

diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CustomKvStateProgram.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/CustomKvStateProgram.java
@@ -32,6 +32,7 @@
 import org.apache.flink.test.util.InfiniteIntegerSource;
 import org.apache.flink.util.Collector;
 
+import java.util.Optional;
 import java.util.concurrent.ThreadLocalRandom;
 
 /**
@@ -47,10 +48,12 @@ public static void main(String[] args) throws Exception {
  final String checkpointPath = args[1];
  final int checkpointingInterval = Integer.parseInt(args[2]);
  final String outputPath = args[3];
+ final Optional<Boolean> unalignedCheckpoints = args.length > 4 ? Optional.of(Boolean.parseBoolean(args[4])) : Optional.empty();
 
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.setParallelism(parallelism);
- env.enableCheckpointing(checkpointingInterval);
+ env.enableCheckpointing(checkpointingInterval);
+ unalignedCheckpoints.ifPresent(value -> env.getCheckpointConfig().enableUnalignedCheckpoints(value));
  env.setStateBackend(new FsStateBackend(checkpointPath));
 
  DataStream<Integer> source = env.addSource(new InfiniteIntegerSource());