XComp commented on code in PR #21736: URL: https://github.com/apache/flink/pull/21736#discussion_r1082750754
########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests Review Comment: ```suggestion * This class aims to generate the snapshots for all the state migration tests. A migration test ``` ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with Review Comment: ```suggestion * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name to match ``` ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); Review Comment: ```suggestion private static final Pattern VERSION_PATTERN = Pattern.compile("v?([0-9]+)[._]([0-9]+)"); ``` ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } + String normalizedVersionName = + "v" + versionMatcher.group(1) + "_" + versionMatcher.group(2); + FlinkVersion version = FlinkVersion.valueOf(normalizedVersionName); + + LOG.info("Start generating for module {} and version {}", rootDirectory, version); + SnapshotGeneratorExecutor executor = new SnapshotGeneratorExecutor(); + List<Class<?>> migrationTests = findMigrationTests(rootDirectory.getAbsolutePath()); + for (Class<?> migrationTestClass : migrationTests) { + LOG.info("Start generating for {}", migrationTestClass.getName()); + executor.executeGenerate(migrationTestClass, version); + LOG.info("Finish generating for {}", migrationTestClass.getName()); Review Comment: ```suggestion LOG.info("Start test data generation for {}", migrationTestClass.getName()); executor.executeGenerate(migrationTestClass, version); LOG.info("Finish test data generation for {}", migrationTestClass.getName()); ``` nit: ...to improve the logs within the Maven build so that it's clear what is generated. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/util/MigrationTest.java: ########## @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.util; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** Interface for state migration tests. */ +public interface MigrationTest { + + /** + * Marks a method as snapshots generator. The method should be like + * + * <pre> + * @@SnapshotsGenerator + * void function(FlinkVersion version) {} + * </pre> + */ + @Retention(RetentionPolicy.RUNTIME) + @Target(ElementType.METHOD) + @interface SnapshotsGenerator {} + + /** + * Marks a method as parameterized snapshots generator. The value should be the method that + * returns Collection of arguments. + * + * <p>The method generating parameters should be like + * + * <pre> + * Collection<?> parameters(FlinkVersion version) {} + * </pre> + * + * <p>The generator method should be like + * + * <pre> + * @@ParameterizedSnapshotsGenerator + * void function(FlinkVersion version, T parameter) {} + * </pre> Review Comment: Isn't that common functionality? Or why do we need to put this in the JavaDoc? :thinking: same goes for the `@SnapshotsGenerator` annotation. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } Review Comment: ```suggestion Preconditions.checkArguments(versionMatcher.matches(), "Version {} could not be parsed. Please specify the version in one of the following formats: {}", versionName, VERSION_PATTERN); ``` it would be nice to have a dedicated flink-utils package that contains flink-independent utility functionality like Preconditions which we then could also use here. I created FLINK-30774 to cover this topic. ########## flink-test-utils-parent/flink-migration-test-utils/pom.xml: ########## @@ -0,0 +1,48 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd";> + + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils-parent</artifactId> + <version>1.17-SNAPSHOT</version> + </parent> + + <artifactId>fink-migration-test-utils</artifactId> Review Comment: What are the reasons for introducing a dedicated module here instead of creating a package in `flink-test-utils` or even `flink-test-utils-junit`? :thinking: I have no strong opinion about it besides keeping the module count low for complexity reasons. I'm just curious about your opinion. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/SnapshotGeneratorExecutor.java: ########## @@ -0,0 +1,188 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.junit.ClassRule; +import org.junit.Rule; Review Comment: I think we should also already come up with a JUnit5 implementation (or, at least, with a follow-up ticket on that one) since JUnit5 is the defacto standard for Flink tests. It would also lower the threshold for contributors to migrate the affected tests to JUnit5 ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobWBroadcastStateMigrationITCase.java: ########## @@ -44,73 +45,80 @@ import org.junit.runners.Parameterized; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; import java.util.Map; -import java.util.stream.Collectors; +import java.util.Optional; /** * Migration ITCases for a stateful job with broadcast state. The tests are parameterized to * (potentially) cover migrating for multiple previous Flink versions, as well as for different * state backends. */ @RunWith(Parameterized.class) -public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { private static final int NUM_SOURCE_ELEMENTS = 4; - // TODO increase this to newer version to create and test snapshot migration for newer versions - private static final FlinkVersion currentVersion = FlinkVersion.v1_16; - - // TODO change this to CREATE_SNAPSHOT to (re)create binary snapshots - // TODO Note: You should generate the snapshot based on the release branch instead of the - // master. - private static final ExecutionMode executionMode = ExecutionMode.VERIFY_SNAPSHOT; - @Parameterized.Parameters(name = "Test snapshot: {0}") public static Collection<SnapshotSpec> parameters() { + return internalParameters(ExecutionMode.VERIFY_SNAPSHOT, Optional.empty()); + } + + public static Collection<SnapshotSpec> internalParameters( + ExecutionMode executionMode, Optional<FlinkVersion> targetGeneratingVersion) { Review Comment: ```suggestion @Nullable FlinkVersion targetGeneratingVersion) { ``` `executionMode` seems to be obsolete here, as well ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobWBroadcastStateMigrationITCase.java: ########## @@ -44,73 +45,80 @@ import org.junit.runners.Parameterized; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; import java.util.Map; -import java.util.stream.Collectors; +import java.util.Optional; /** * Migration ITCases for a stateful job with broadcast state. The tests are parameterized to * (potentially) cover migrating for multiple previous Flink versions, as well as for different * state backends. */ @RunWith(Parameterized.class) -public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { private static final int NUM_SOURCE_ELEMENTS = 4; - // TODO increase this to newer version to create and test snapshot migration for newer versions - private static final FlinkVersion currentVersion = FlinkVersion.v1_16; - - // TODO change this to CREATE_SNAPSHOT to (re)create binary snapshots - // TODO Note: You should generate the snapshot based on the release branch instead of the - // master. - private static final ExecutionMode executionMode = ExecutionMode.VERIFY_SNAPSHOT; - @Parameterized.Parameters(name = "Test snapshot: {0}") public static Collection<SnapshotSpec> parameters() { Review Comment: ```suggestion public static Collection<SnapshotSpec> createSpecsForTestRuns() { ``` ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -60,62 +62,68 @@ * previous Flink versions, as well as for different state backends. */ @RunWith(Parameterized.class) -public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { private static final int NUM_SOURCE_ELEMENTS = 4; - // TODO increase this to newer version to create and test snapshot migration for newer versions - private static final FlinkVersion currentVersion = FlinkVersion.v1_16; - - // TODO change this to CREATE_SNAPSHOT to (re)create binary snapshots - // TODO Note: You should generate the snapshot based on the release branch instead of the - // master. - private static final ExecutionMode executionMode = ExecutionMode.VERIFY_SNAPSHOT; - @Parameterized.Parameters(name = "Test snapshot: {0}") public static Collection<SnapshotSpec> parameters() { Review Comment: ```suggestion public static Collection<SnapshotSpec> createSpecsForTestRuns() { ``` ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -60,62 +62,68 @@ * previous Flink versions, as well as for different state backends. */ @RunWith(Parameterized.class) -public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { Review Comment: This applies to the other tests as well. I won't mentioned it there. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } + String normalizedVersionName = + "v" + versionMatcher.group(1) + "_" + versionMatcher.group(2); + FlinkVersion version = FlinkVersion.valueOf(normalizedVersionName); + + LOG.info("Start generating for module {} and version {}", rootDirectory, version); + SnapshotGeneratorExecutor executor = new SnapshotGeneratorExecutor(); + List<Class<?>> migrationTests = findMigrationTests(rootDirectory.getAbsolutePath()); + for (Class<?> migrationTestClass : migrationTests) { + LOG.info("Start generating for {}", migrationTestClass.getName()); + executor.executeGenerate(migrationTestClass, version); + LOG.info("Finish generating for {}", migrationTestClass.getName()); + } + + // Avoids leaking threads blocks the process. + System.exit(0); Review Comment: I feel like we should not try to work around issues within the code here but rather fix these issues in the code. If we have threads not terminating properly. WDYT? ########## flink-annotations/src/main/java/org/apache/flink/FlinkVersion.java: ########## @@ -86,6 +86,10 @@ public static Optional<FlinkVersion> byCode(String code) { return Optional.ofNullable(CODE_MAP.get(code)); } + public static FlinkVersion last() { + return values()[values().length - 2]; + } + /** Returns the current version. */ public static FlinkVersion current() { Review Comment: ```suggestion public static FlinkVersion getMostRecentlyPublishedVersion() { return values()[values().length - 2]; } /** Returns the current version. */ public static FlinkVersion getCurrentMasterVersion() { ``` Can we make these method names more descriptive? `current` couldn't can stand for different versions (e.g. published vs not-published) ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/util/MigrationTest.java: ########## @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.util; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** Interface for state migration tests. */ +public interface MigrationTest { + + /** + * Marks a method as snapshots generator. The method should be like + * + * <pre> + * @@SnapshotsGenerator Review Comment: ```suggestion * @SnapshotsGenerator ``` ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/SnapshotGeneratorExecutor.java: ########## @@ -0,0 +1,188 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.rules.ExternalResource; +import org.junit.rules.RunRules; +import org.junit.rules.TestRule; +import org.junit.runner.Description; +import org.junit.runners.Parameterized; +import org.junit.runners.model.Statement; + +import java.lang.reflect.Array; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +/** A utility class to execute generateSnapshots method for migration tests. */ +public class SnapshotGeneratorExecutor { + + public void executeGenerate(Class<?> migrationTestClass, FlinkVersion flinkVersion) + throws Throwable { + for (Method method : migrationTestClass.getMethods()) { + if (method.isAnnotationPresent(MigrationTest.SnapshotsGenerator.class)) { + executeGenerateMethods(migrationTestClass, method, flinkVersion); + } else if (method.isAnnotationPresent( + MigrationTest.ParameterizedSnapshotsGenerator.class)) { + String parametersMethodName = + method.getAnnotation(MigrationTest.ParameterizedSnapshotsGenerator.class) + .value(); + Method parametersMethod = + migrationTestClass.getMethod(parametersMethodName, FlinkVersion.class); + executeParameterizedGenerateMethods( + migrationTestClass, method, parametersMethod, flinkVersion); + } + } + } + + private void executeGenerateMethods( + Class<?> migrationTestClass, Method method, FlinkVersion version) throws Throwable { + method.setAccessible(true); + List<TestRule> classRules = getClassRules(migrationTestClass); + + executeWithRules( + classRules, + new Statement() { + @Override + public void evaluate() throws Throwable { + Object migrationTest = createMigrationTest(migrationTestClass); + List<TestRule> rules = getRules(migrationTest); + executeWithRules( + rules, + new Statement() { + @Override + public void evaluate() throws Throwable { + method.invoke(migrationTest, version); + } + }); + } + }); + } + + private void executeParameterizedGenerateMethods( + Class<?> migrationTestClass, + Method method, + Method parametersMethod, + FlinkVersion version) + throws Throwable { + method.setAccessible(true); + parametersMethod.setAccessible(true); + List<TestRule> classRules = getClassRules(migrationTestClass); + + executeWithRules( + classRules, + new Statement() { + @Override + public void evaluate() throws Throwable { + Object migrationTest = createMigrationTest(migrationTestClass); + List<TestRule> rules = getRules(migrationTest); + Collection<?> arguments = + (Collection<?>) parametersMethod.invoke(migrationTest, version); + for (Object argument : arguments) { + executeWithRules( + rules, + new Statement() { + @Override + public void evaluate() throws Throwable { + method.invoke(migrationTest, version, argument); + } + }); + } + } + }); + } + + private void executeWithRules(List<TestRule> rules, Statement statement) throws Throwable { + new RunRules(statement, rules, Description.EMPTY).evaluate(); + } + + private List<TestRule> getClassRules(Class<?> migrationTestClass) + throws IllegalAccessException { + List<TestRule> classRules = new ArrayList<>(); + + Field[] fields = migrationTestClass.getFields(); + for (Field field : fields) { + if (ExternalResource.class.isAssignableFrom(field.getType()) + && field.isAnnotationPresent(ClassRule.class)) { + field.setAccessible(true); + classRules.add((ExternalResource) field.get(null)); + } + } + + return classRules; + } + + private static Object createMigrationTest(Class<?> migrationTestClass) throws Exception { + Constructor<?>[] constructors = migrationTestClass.getDeclaredConstructors(); + for (Constructor<?> constructor : constructors) { + if (constructor.getParameterCount() == 0) { + constructor.setAccessible(true); + return constructor.newInstance(); + } + } + + // This class does not have a constructor without argument. + Constructor<?> constructor = constructors[0]; + constructor.setAccessible(true); + + // Check if we could find method labeled with @Parameterized.Parameters + for (Method method : migrationTestClass.getMethods()) { + if (Modifier.isStatic(method.getModifiers()) + && method.isAnnotationPresent(Parameterized.Parameters.class)) { + Object argumentLists = method.invoke(null); + if (argumentLists instanceof Collection) { + return constructor.newInstance( + ((Collection<?>) argumentLists).iterator().next()); + } else if (argumentLists.getClass().isArray()) { + return constructor.newInstance(Array.get(argumentLists, 0)); + } else { + throw new RuntimeException( + "Failed to create parameterized class object due to argument lists type not supported: " + + argumentLists.getClass()); + } + } + } + + throw new RuntimeException("Could not create the object for " + migrationTestClass); + } + + private List<TestRule> getRules(Object migrationTest) throws IllegalAccessException { Review Comment: `getRules` and `getClassRules` are almost the same. We could merge them into a single method. Additionally, that could become a static method as far as I can see. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } + String normalizedVersionName = + "v" + versionMatcher.group(1) + "_" + versionMatcher.group(2); + FlinkVersion version = FlinkVersion.valueOf(normalizedVersionName); Review Comment: We could move this logic into `FlinkVersion` providing a method `valueOf(int majorVersion, int minorVersion)`. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } + String normalizedVersionName = + "v" + versionMatcher.group(1) + "_" + versionMatcher.group(2); + FlinkVersion version = FlinkVersion.valueOf(normalizedVersionName); + + LOG.info("Start generating for module {} and version {}", rootDirectory, version); + SnapshotGeneratorExecutor executor = new SnapshotGeneratorExecutor(); + List<Class<?>> migrationTests = findMigrationTests(rootDirectory.getAbsolutePath()); + for (Class<?> migrationTestClass : migrationTests) { + LOG.info("Start generating for {}", migrationTestClass.getName()); + executor.executeGenerate(migrationTestClass, version); + LOG.info("Finish generating for {}", migrationTestClass.getName()); + } + + // Avoids leaking threads blocks the process. + System.exit(0); + } catch (Throwable e) { + LOG.error("Failed to generate snapshots for the state migration tests", e); + throw new RuntimeException(e); + } + } + + private static List<Class<?>> findMigrationTests(String rootDirectory) throws Exception { + List<Class<?>> migrationTests = new ArrayList<>(); + for (String prefix : DEFAULT_PATH_PREFIXES) { Review Comment: It feels like pathPrefixes should be some kind of parameter as well. `DEFAULT_PATH_PREFIXES` could be the fallback if this parameter is not set. ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/MigrationTestsSnapshotGenerator.java: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class aims to generate the snapshots for all the state migration tests. A migration tests + * should implement {@link MigrationTest} interface and its name should match {@code + * *(Test|ITCase)(.java|.scala)}. For scala tests, we also require the class name is the same with + * the containing filename. + */ +public class MigrationTestsSnapshotGenerator { + + private static final Logger LOG = + LoggerFactory.getLogger(MigrationTestsSnapshotGenerator.class); + + private static final String[] DEFAULT_PATH_PREFIXES = + new String[] {"src/test/java", "src/test/scala"}; + + private static final Pattern VERSION_PATTERN = Pattern.compile("" + "v?([0-9]+)[._]([0-9]+)"); + + private static final String CLASS_NAME_GROUP = "className"; + private static final Pattern CLASS_NAME_PATTERN = + Pattern.compile("(?<" + CLASS_NAME_GROUP + ">[a-zA-Z0-9]*(Test|ITCase))(.java|.scala)"); + + public static void main(String[] args) { + try { + if (args.length < 2) { + throw new IllegalArgumentException( + "Usage: java MigrationTestsSnapshotGenerator [project root path] [version]"); + } + + File rootDirectory = new File(args[0]); + if (!rootDirectory.exists() || !rootDirectory.isDirectory()) { + throw new FileNotFoundException( + rootDirectory + " does not exist or is not a directory."); + } + + String versionName = args[1]; + Matcher versionMatcher = VERSION_PATTERN.matcher(versionName); + if (!versionMatcher.matches()) { + throw new IllegalArgumentException( + "Version " + + versionName + + "could not be parsed, " + + "please specify the version with format like 1.17, 1_17, v1_17, v1.17"); + } + String normalizedVersionName = + "v" + versionMatcher.group(1) + "_" + versionMatcher.group(2); + FlinkVersion version = FlinkVersion.valueOf(normalizedVersionName); + + LOG.info("Start generating for module {} and version {}", rootDirectory, version); + SnapshotGeneratorExecutor executor = new SnapshotGeneratorExecutor(); + List<Class<?>> migrationTests = findMigrationTests(rootDirectory.getAbsolutePath()); + for (Class<?> migrationTestClass : migrationTests) { + LOG.info("Start generating for {}", migrationTestClass.getName()); + executor.executeGenerate(migrationTestClass, version); + LOG.info("Finish generating for {}", migrationTestClass.getName()); + } + + // Avoids leaking threads blocks the process. + System.exit(0); + } catch (Throwable e) { + LOG.error("Failed to generate snapshots for the state migration tests", e); + throw new RuntimeException(e); + } + } + + private static List<Class<?>> findMigrationTests(String rootDirectory) throws Exception { + List<Class<?>> migrationTests = new ArrayList<>(); + for (String prefix : DEFAULT_PATH_PREFIXES) { + recursivelyFindMigrationTests(rootDirectory, prefix, "", migrationTests); + } + + return migrationTests; + } + + private static void recursivelyFindMigrationTests( Review Comment: Is there a reason why we don't work with return values here? ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/migration/SnapshotGeneratorExecutor.java: ########## @@ -0,0 +1,188 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.migration; + +import org.apache.flink.FlinkVersion; +import org.apache.flink.test.util.MigrationTest; + +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.rules.ExternalResource; +import org.junit.rules.RunRules; +import org.junit.rules.TestRule; +import org.junit.runner.Description; +import org.junit.runners.Parameterized; +import org.junit.runners.model.Statement; + +import java.lang.reflect.Array; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +/** A utility class to execute generateSnapshots method for migration tests. */ +public class SnapshotGeneratorExecutor { + + public void executeGenerate(Class<?> migrationTestClass, FlinkVersion flinkVersion) Review Comment: ```suggestion class SnapshotGeneratorExecutor { void executeGenerate(Class<?> migrationTestClass, FlinkVersion flinkVersion) ``` both can be made package-private ########## flink-test-utils-parent/flink-migration-test-utils/src/main/java/org/apache/flink/test/util/MigrationTest.java: ########## @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.util; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** Interface for state migration tests. */ +public interface MigrationTest { + + /** + * Marks a method as snapshots generator. The method should be like + * + * <pre> + * @@SnapshotsGenerator + * void function(FlinkVersion version) {} + * </pre> + */ + @Retention(RetentionPolicy.RUNTIME) + @Target(ElementType.METHOD) + @interface SnapshotsGenerator {} + + /** + * Marks a method as parameterized snapshots generator. The value should be the method that + * returns Collection of arguments. + * + * <p>The method generating parameters should be like + * + * <pre> + * Collection<?> parameters(FlinkVersion version) {} + * </pre> + * + * <p>The generator method should be like + * + * <pre> + * @@ParameterizedSnapshotsGenerator Review Comment: ```suggestion * @ParameterizedSnapshotsGenerator ``` ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobWBroadcastStateMigrationITCase.java: ########## @@ -44,73 +45,80 @@ import org.junit.runners.Parameterized; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; import java.util.Map; -import java.util.stream.Collectors; +import java.util.Optional; /** * Migration ITCases for a stateful job with broadcast state. The tests are parameterized to * (potentially) cover migrating for multiple previous Flink versions, as well as for different * state backends. */ @RunWith(Parameterized.class) -public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobWBroadcastStateMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { private static final int NUM_SOURCE_ELEMENTS = 4; - // TODO increase this to newer version to create and test snapshot migration for newer versions - private static final FlinkVersion currentVersion = FlinkVersion.v1_16; - - // TODO change this to CREATE_SNAPSHOT to (re)create binary snapshots - // TODO Note: You should generate the snapshot based on the release branch instead of the - // master. - private static final ExecutionMode executionMode = ExecutionMode.VERIFY_SNAPSHOT; - @Parameterized.Parameters(name = "Test snapshot: {0}") public static Collection<SnapshotSpec> parameters() { + return internalParameters(ExecutionMode.VERIFY_SNAPSHOT, Optional.empty()); + } + + public static Collection<SnapshotSpec> internalParameters( Review Comment: ```suggestion private static Collection<SnapshotSpec> internalParameters( ``` ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -125,8 +133,23 @@ public StatefulJobSnapshotMigrationITCase(SnapshotSpec snapshotSpec) throws Exce this.snapshotSpec = snapshotSpec; } + public static Collection<SnapshotSpec> generateParameters(FlinkVersion targetVersion) { Review Comment: ```suggestion public static Collection<SnapshotSpec> createSpecsForTestDataGeneration(FlinkVersion targetVersion) { ``` `generateParameters` and `parameters` are quite similar method names. We might want to clarify their purpose in the method name properly. Additionally, could we move this method underneath the `parameters()` method to show that they are closely related? ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -60,62 +62,68 @@ * previous Flink versions, as well as for different state backends. */ @RunWith(Parameterized.class) -public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { Review Comment: Why do we add the interface here? Isn't `MigrationTest` not only used for annotations? No methods are introduced through this interface AFAIS. :thinking: ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -60,62 +62,68 @@ * previous Flink versions, as well as for different state backends. */ @RunWith(Parameterized.class) -public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase { +public class StatefulJobSnapshotMigrationITCase extends SnapshotMigrationTestBase + implements MigrationTest { private static final int NUM_SOURCE_ELEMENTS = 4; - // TODO increase this to newer version to create and test snapshot migration for newer versions - private static final FlinkVersion currentVersion = FlinkVersion.v1_16; - - // TODO change this to CREATE_SNAPSHOT to (re)create binary snapshots - // TODO Note: You should generate the snapshot based on the release branch instead of the - // master. - private static final ExecutionMode executionMode = ExecutionMode.VERIFY_SNAPSHOT; - @Parameterized.Parameters(name = "Test snapshot: {0}") public static Collection<SnapshotSpec> parameters() { + return internalParameters(ExecutionMode.VERIFY_SNAPSHOT, Optional.empty()); + } + + private static Collection<SnapshotSpec> internalParameters( + ExecutionMode executionMode, Optional<FlinkVersion> targetGeneratingVersion) { Review Comment: ```suggestion private static Collection<SnapshotSpec> internalParameters(@Nullable FlinkVersion targetGeneratingVersion) { ``` It looks like the `executionMode` is kind of redundant in this method. We rely on `targetGeneratingVersion` if it's set. Otherwise, we have a default version collection that's defined when creating the individual specs. ########## flink-tests/src/test/java/org/apache/flink/test/checkpointing/StatefulJobSnapshotMigrationITCase.java: ########## @@ -125,8 +133,23 @@ public StatefulJobSnapshotMigrationITCase(SnapshotSpec snapshotSpec) throws Exce this.snapshotSpec = snapshotSpec; } + public static Collection<SnapshotSpec> generateParameters(FlinkVersion targetVersion) { Review Comment: ```suggestion private static Collection<SnapshotSpec> generateParameters(FlinkVersion targetVersion) { ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
