[FLINK-13159] [tests] Add subclass serialization cases to PojoSeriali…

…zerSnapshotMigrationTest

This closes apache#9375.

flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotMigrationTest.java

@@ -31,6 +31,7 @@
 import java.util.Collection;
 
 import static org.apache.flink.api.common.typeutils.TypeSerializerMatchers.hasSameCompatibilityAs;
+import static org.apache.flink.api.common.typeutils.TypeSerializerMatchers.isCompatibleWithReconfiguredSerializer;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -84,6 +85,22 @@ public TestPojoWithNewAndRemovedFields(String name, int age, String githubId, Co
  }
  }
 
+ /**
+ * Data test files of test specification that used this type
+ * had the data written using {@code PojoSerializer}s generated
+ * using the base class {@code TestPojo}.
+ */
+ public static class TestPojoSubclass extends TestPojo {
+ public String githubId;
+
+ public TestPojoSubclass() {}
+
+ public TestPojoSubclass(int id, String name, int age, String githubId) {
+ super(id, name, age);
+ this.githubId = githubId;
+ }
+ }
+
  public PojoSerializerSnapshotMigrationTest(TestSpecification<Object> testSpecification) {
  super(testSpecification);
  }
@@ -107,6 +124,19 @@ public static Collection<TestSpecification<?>> testSpecifications() {
  PojoSerializerSnapshotMigrationTest::testPojoWithNewAndRemovedFieldsSerializerSupplier,
  hasSameCompatibilityAs(TypeSerializerSchemaCompatibility.compatibleAfterMigration()));
 
+ testSpecifications.addWithCompatibilityMatcher(
+ "pojo-unregistered-subclass-serializer",
+ PojoSerializer.class,
+ PojoSerializerSnapshot.class,
+ PojoSerializerSnapshotMigrationTest::testPojoSerializerSupplier,
+ isCompatibleWithReconfiguredSerializer());
+
+ testSpecifications.add(
+ "pojo-registered-subclass-serializer",
+ PojoSerializer.class,
+ PojoSerializerSnapshot.class,
+ PojoSerializerSnapshotMigrationTest::testPojoSerializerWithSubclassRegisteredSupplier);
+
  return testSpecifications.get();
  }
 
@@ -125,4 +155,15 @@ private static TypeSerializer<TestPojoWithNewAndRemovedFields> testPojoWithNewAn
  assertTrue(serializer instanceof PojoSerializer);
  return serializer;
  }
+
+ private static TypeSerializer<TestPojo> testPojoSerializerWithSubclassRegisteredSupplier() {
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.registerPojoType(TestPojoSubclass.class);
+
+ TypeSerializer<TestPojo> serializer =
+ TypeExtractor.createTypeInfo(TestPojo.class).createSerializer(executionConfig);
+
+ assertTrue(serializer instanceof PojoSerializer);
+ return serializer;
+ }
 }