[FLINK-37404, FLINK-36334][API / Core] Reenable Migration Tests for Filnk 2.x

flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/VariantSerializerUpgradeTest.java

@@ -29,17 +29,11 @@
 import org.apache.flink.types.variant.VariantBuilder;
 
 import org.assertj.core.api.Condition;
-import org.junit.jupiter.api.Disabled;
 
 import java.util.ArrayList;
 import java.util.Collection;
 
-/**
- * A {@link TypeSerializerUpgradeTestBase} for {@link VariantSerializerSnapshot}. The test is
- * disabled because Variant is introduced in Flink 2.1. We should restore the test when there is a
- * Flink 2.2 which should test compatibility with Flink 2.1
- */
-@Disabled("FLINK-37951")
+/** A {@link TypeSerializerUpgradeTestBase} for {@link VariantSerializerSnapshot}. */
 class VariantSerializerUpgradeTest extends TypeSerializerUpgradeTestBase<Variant, Variant> {
 
     private static final String SPEC_NAME = "variant-serializer";

flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java

@@ -36,7 +36,6 @@
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
 import org.apache.flink.test.util.MigrationTest;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -54,14 +53,9 @@
  * Tests for checking whether CEP operator can restore from snapshots that were done using previous
  * Flink versions.
  *
- * <p>NOTE: Due to major upgrades of serializers, Flink 2.0 is not compatible with saved state from
- * Flink 1.x, and this test is temporarily disabled. This test should be reinstated for future
- * versions of Flink, such as Flink 2.1, that promise compatibility with older releases.
- *
  * <p>For regenerating the binary snapshot file of previous versions you have to run the {@code
  * write*()} method on the corresponding Flink release-* branch.
  */
-@Ignore("Flink 2.0 is not compatible with saved state from Flink 1.x")
 @RunWith(Parameterized.class)
 public class CEPMigrationTest implements MigrationTest {
 
@@ -70,7 +64,7 @@ public class CEPMigrationTest implements MigrationTest {
     @Parameterized.Parameters(name = "Migration Savepoint: {0}")
     public static Collection<FlinkVersion> parameters() {
         return FlinkVersion.rangeOf(
-                FlinkVersion.v1_20, MigrationTest.getMostRecentlyPublishedVersion());
+                FlinkVersion.v2_0, MigrationTest.getMostRecentlyPublishedVersion());
     }
 
     public CEPMigrationTest(FlinkVersion migrateVersion) {

flink-table/flink-table-api-scala/src/test/java/org/apache/flink/table/api/typeutils/OptionSerializerUpgradeTest.java

@@ -27,7 +27,6 @@
 import org.apache.flink.test.util.MigrationTest;
 
 import org.assertj.core.api.Condition;
-import org.junit.jupiter.api.Disabled;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -38,7 +37,6 @@
  * A {@link org.apache.flink.api.common.typeutils.TypeSerializerUpgradeTestBase} for {@link
  * ScalaEitherSerializerSnapshot}.
  */
-@Disabled("FLINK-36334")
 class OptionSerializerUpgradeTest
         extends TypeSerializerUpgradeTestBase<Option<String>, Option<String>> {
 

flink-table/flink-table-api-scala/src/test/java/org/apache/flink/table/api/typeutils/ScalaEitherSerializerUpgradeTest.java

@@ -28,7 +28,6 @@
 import org.apache.flink.test.util.MigrationTest;
 
 import org.assertj.core.api.Condition;
-import org.junit.jupiter.api.Disabled;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -37,7 +36,6 @@
 import scala.util.Right;
 
 /** A {@link TypeSerializerUpgradeTestBase} for {@link ScalaEitherSerializerSnapshot}. */
-@Disabled("FLINK-36334")
 class ScalaEitherSerializerUpgradeTest
         extends TypeSerializerUpgradeTestBase<Either<Integer, String>, Either<Integer, String>> {
 

flink-table/flink-table-api-scala/src/test/java/org/apache/flink/table/api/typeutils/ScalaTrySerializerUpgradeTest.java

@@ -28,7 +28,6 @@
 import org.apache.flink.test.util.MigrationTest;
 
 import org.assertj.core.api.Condition;
-import org.junit.jupiter.api.Disabled;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -38,7 +37,6 @@
 import scala.util.Try;
 
 /** A {@link TypeSerializerUpgradeTestBase} for {@link TrySerializer}. */
-@Disabled("FLINK-36334")
 class ScalaTrySerializerUpgradeTest
         extends TypeSerializerUpgradeTestBase<Try<String>, Try<String>> {
 

flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/typeutils/EnumValueSerializerUpgradeTest.scala

@@ -24,13 +24,11 @@ import org.apache.flink.table.api.typeutils.EnumValueSerializerUpgradeTest.{Enum
 import org.apache.flink.test.util.MigrationTest
 
 import org.assertj.core.api.Condition
-import org.junit.jupiter.api.Disabled
 
 import java.util
 import java.util.Objects
 
 /** A [[TypeSerializerUpgradeTestBase]] for [[EnumValueSerializer]]. */
-@Disabled("FLINK-36334")
 class EnumValueSerializerUpgradeTest
   extends TypeSerializerUpgradeTestBase[Letters.Value, Letters.Value] {
 

flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/typeutils/ScalaCaseClassSerializerUpgradeTest.scala

@@ -27,12 +27,10 @@ import org.apache.flink.table.types.CustomCaseClass
 import org.apache.flink.test.util.MigrationTest
 
 import org.assertj.core.api.Condition
-import org.junit.jupiter.api.Disabled
 
 import java.util
 
 /** A [[TypeSerializerUpgradeTestBase]] for [[ScalaCaseClassSerializer]]. */
-@Disabled("FLINK-36334")
 class ScalaCaseClassSerializerUpgradeTest
   extends TypeSerializerUpgradeTestBase[CustomCaseClass, CustomCaseClass] {