[FLINK-35306][test] Migrate flink-tests-java17 module to JUnit5 and assertj to solve jdk17 compile fails

Author: 1996fanrui

flink-tests-java17/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoRecordSerializerUpgradeTestSpecifications.java

@@ -27,10 +27,9 @@
 import org.apache.flink.api.common.typeutils.TypeSerializerUpgradeTestBase;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 
-import org.hamcrest.Matcher;
+import org.assertj.core.api.Condition;
 
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertSame;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** A {@link TypeSerializerUpgradeTestBase} for the {@link PojoSerializer}. */
 class PojoRecordSerializerUpgradeTestSpecifications {
@@ -58,7 +57,7 @@ public TypeSerializer<PojoBeforeUpgrade> createPriorSerializer() {
             TypeSerializer<PojoBeforeUpgrade> serializer =
                     TypeExtractor.createTypeInfo(PojoBeforeUpgrade.class)
                             .createSerializer(new SerializerConfigImpl());
-            assertSame(PojoSerializer.class, serializer.getClass());
+            assertThat(serializer.getClass()).isSameAs(PojoSerializer.class);
             return serializer;
         }
 
@@ -81,18 +80,19 @@ public TypeSerializer<PojoAfterUpgrade> createUpgradedSerializer() {
             TypeSerializer<PojoAfterUpgrade> serializer =
                     TypeExtractor.createTypeInfo(PojoAfterUpgrade.class)
                             .createSerializer(new SerializerConfigImpl());
-            assertSame(PojoSerializer.class, serializer.getClass());
+            assertThat(serializer.getClass()).isSameAs(PojoSerializer.class);
             return serializer;
         }
 
         @Override
-        public Matcher<PojoAfterUpgrade> testDataMatcher() {
-            return is(new PojoAfterUpgrade(911108, "Gordon"));
+        public Condition<PojoAfterUpgrade> testDataCondition() {
+            return new Condition<>(
+                    new PojoAfterUpgrade(911108, "Gordon")::equals, "value is (911108, Gordon)");
         }
 
         @Override
-        public Matcher<TypeSerializerSchemaCompatibility<PojoAfterUpgrade>>
-                schemaCompatibilityMatcher(FlinkVersion version) {
+        public Condition<TypeSerializerSchemaCompatibility<PojoAfterUpgrade>>
+                schemaCompatibilityCondition(FlinkVersion version) {
             return TypeSerializerConditions.isCompatibleAsIs();
         }
     }
@@ -110,7 +110,7 @@ public TypeSerializer<RecordBeforeMigration> createPriorSerializer() {
             TypeSerializer<RecordBeforeMigration> serializer =
                     TypeExtractor.createTypeInfo(RecordBeforeMigration.class)
                             .createSerializer(new SerializerConfigImpl());
-            assertSame(PojoSerializer.class, serializer.getClass());
+            assertThat(serializer.getClass()).isSameAs(PojoSerializer.class);
             return serializer;
         }
 
@@ -133,18 +133,20 @@ public TypeSerializer<RecordAfterSchemaUpgrade> createUpgradedSerializer() {
             TypeSerializer<RecordAfterSchemaUpgrade> serializer =
                     TypeExtractor.createTypeInfo(RecordAfterSchemaUpgrade.class)
                             .createSerializer(new SerializerConfigImpl());
-            assertSame(PojoSerializer.class, serializer.getClass());
+            assertThat(serializer.getClass()).isSameAs(PojoSerializer.class);
             return serializer;
         }
 
         @Override
-        public Matcher<RecordAfterSchemaUpgrade> testDataMatcher() {
-            return is(new RecordAfterSchemaUpgrade("Gordon", 0, null));
+        public Condition<RecordAfterSchemaUpgrade> testDataCondition() {
+            return new Condition<>(
+                    new RecordAfterSchemaUpgrade("Gordon", 0, null)::equals,
+                    "value is (Gordon, 0 ,null)");
         }
 
         @Override
-        public Matcher<TypeSerializerSchemaCompatibility<RecordAfterSchemaUpgrade>>
-                schemaCompatibilityMatcher(FlinkVersion version) {
+        public Condition<TypeSerializerSchemaCompatibility<RecordAfterSchemaUpgrade>>
+                schemaCompatibilityCondition(FlinkVersion version) {
             return TypeSerializerConditions.isCompatibleAfterMigration();
         }
     }