You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2018/10/30 13:57:19 UTC
[flink] 12/16: [FLINK-10709] [core] Remove NS generic from
TypeSerializerSchemaCompatibility
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 17f311aadacab2f6b0a19de552017dbfeb0681e1
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Sun Oct 28 16:17:27 2018 +0100
[FLINK-10709] [core] Remove NS generic from TypeSerializerSchemaCompatibility
---
.../BackwardsCompatibleSerializerSnapshot.java | 2 +-
.../flink/api/common/typeutils/CompatibilityUtil.java | 2 +-
.../common/typeutils/CompositeSerializerSnapshot.java | 13 ++++++-------
.../common/typeutils/TypeSerializerConfigSnapshot.java | 4 ++--
.../typeutils/TypeSerializerSchemaCompatibility.java | 18 ++++++++----------
.../api/common/typeutils/TypeSerializerSnapshot.java | 3 +--
.../base/GenericArraySerializerConfigSnapshot.java | 2 +-
.../api/common/typeutils/base/ListSerializer.java | 4 ++--
.../runtime/EitherSerializerConfigSnapshot.java | 10 +++++-----
.../typeutils/runtime/EitherSerializerSnapshot.java | 2 +-
.../flink/api/common/typeutils/SerializerTestBase.java | 2 +-
.../TypeSerializerSnapshotMigrationTestBase.java | 2 +-
.../api/common/typeutils/base/EnumSerializerTest.java | 2 +-
.../api/java/typeutils/runtime/PojoSerializerTest.java | 10 +++++-----
.../runtime/kryo/KryoSerializerCompatibilityTest.java | 6 +++---
.../formats/avro/typeutils/AvroSerializerSnapshot.java | 8 ++++----
.../runtime/state/DefaultOperatorStateBackend.java | 6 +++---
.../runtime/state/heap/HeapKeyedStateBackend.java | 6 +++---
.../runtime/state/StateBackendMigrationTestBase.java | 5 +++--
.../typeutils/EnumValueSerializerUpgradeTest.scala | 2 +-
.../streaming/state/RocksDBKeyedStateBackend.java | 6 +++---
.../TypeSerializerSnapshotMigrationITCase.java | 2 +-
22 files changed, 57 insertions(+), 60 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/BackwardsCompatibleSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/BackwardsCompatibleSerializerSnapshot.java
index 4b72902..18c20e8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/BackwardsCompatibleSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/BackwardsCompatibleSerializerSnapshot.java
@@ -72,7 +72,7 @@ public class BackwardsCompatibleSerializerSnapshot<T> implements TypeSerializerS
}
@Override
- public <NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
// if there is no configuration snapshot to check against,
// then we can only assume that the new serializer is compatible as is
return TypeSerializerSchemaCompatibility.compatibleAsIs();
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompatibilityUtil.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompatibilityUtil.java
index ae19d03..34ca702 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompatibilityUtil.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompatibilityUtil.java
@@ -95,7 +95,7 @@ public class CompatibilityUtil {
TypeSerializerSnapshot<T> precedingSerializerConfigSnapshot,
TypeSerializer<T> newSerializer) {
- TypeSerializerSchemaCompatibility<T, TypeSerializer<T>> compatibility =
+ TypeSerializerSchemaCompatibility<T> compatibility =
precedingSerializerConfigSnapshot.resolveSchemaCompatibility(newSerializer);
// everything except "compatible" maps to "requires migration".
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializerSnapshot.java
index 473a203..982a7ee 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializerSnapshot.java
@@ -76,7 +76,7 @@ public abstract class CompositeSerializerSnapshot<T, S extends TypeSerializer<T>
protected abstract TypeSerializer<?>[] getNestedSerializersFromSerializer(S serializer);
- protected abstract TypeSerializerSchemaCompatibility<T, S> outerCompatibility(S serializer);
+ protected abstract TypeSerializerSchemaCompatibility<T> outerCompatibility(S serializer);
protected abstract Class<?> outerSerializerType();
@@ -140,8 +140,7 @@ public abstract class CompositeSerializerSnapshot<T, S extends TypeSerializer<T>
return createSerializer(nestedSerializers);
}
- public <NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS>
- resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
// class compatibility
if (!outerSerializerType().isInstance(newSerializer)) {
@@ -152,7 +151,7 @@ public abstract class CompositeSerializerSnapshot<T, S extends TypeSerializer<T>
@SuppressWarnings("unchecked")
final S castedSerializer = (S) newSerializer;
- final TypeSerializerSchemaCompatibility<T, S> outerCompatibility = outerCompatibility(castedSerializer);
+ final TypeSerializerSchemaCompatibility<T> outerCompatibility = outerCompatibility(castedSerializer);
if (outerCompatibility.isIncompatible()) {
return TypeSerializerSchemaCompatibility.incompatible();
@@ -165,7 +164,7 @@ public abstract class CompositeSerializerSnapshot<T, S extends TypeSerializer<T>
boolean nestedSerializerRequiresMigration = false;
for (int i = 0; i < nestedSnapshots.length; i++) {
- TypeSerializerSchemaCompatibility<?, ?> compatibility =
+ TypeSerializerSchemaCompatibility<?> compatibility =
resolveCompatibility(nestedSerializers[i], nestedSnapshots[i]);
if (compatibility.isIncompatible()) {
@@ -189,11 +188,11 @@ public abstract class CompositeSerializerSnapshot<T, S extends TypeSerializer<T>
* Utility method to conjure up a new scope for the generic parameters.
*/
@SuppressWarnings("unchecked")
- private static <E, X extends TypeSerializer<E>> TypeSerializerSchemaCompatibility<E, X> resolveCompatibility(
+ private static <E> TypeSerializerSchemaCompatibility<E> resolveCompatibility(
TypeSerializer<?> serializer,
TypeSerializerSnapshot<?> snapshot) {
- X typedSerializer = (X) serializer;
+ TypeSerializer<E> typedSerializer = (TypeSerializer<E>) serializer;
TypeSerializerSnapshot<E> typedSnapshot = (TypeSerializerSnapshot<E>) snapshot;
return typedSnapshot.resolveSchemaCompatibility(typedSerializer);
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
index 5be3987..53d2cef 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
@@ -144,8 +144,8 @@ public abstract class TypeSerializerConfigSnapshot<T> extends VersionedIOReadabl
}
@Override
- public <NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> resolveSchemaCompatibility(
- NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
// in prior versions, the compatibility check was in the serializer itself, so we
// delegate this call to the serializer.
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSchemaCompatibility.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSchemaCompatibility.java
index 6555f9b..d488799 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSchemaCompatibility.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSchemaCompatibility.java
@@ -27,20 +27,18 @@ import javax.annotation.Nullable;
* A {@code TypeSerializerSchemaCompatibility} represents information about whether or not a {@link TypeSerializer}
* can be safely used to read data written by a previous type serializer.
*
- * <p>Typically, the compatibility of the new serializer is resolved by checking it against the snapshotted
- * {@link TypeSerializerConfigSnapshot} of the previous serializer. Depending on the type of the
+ * <p>Typically, the compatibility of the new serializer is resolved by checking the serializer against the
+ * {@link TypeSerializerSnapshot} of the previous serializer. Depending on the type of the
* resolved compatibility result, migration (i.e., reading bytes with the previous serializer and then writing
* it again with the new serializer) may be required before the new serializer can be used.
*
* @param <T> the type of data serialized by the serializer that was being checked.
*
- * @param <NS> the type of serializer that was being checked.
- *
* @see TypeSerializer
- * @see TypeSerializerConfigSnapshot#resolveSchemaCompatibility(TypeSerializer)
+ * @see TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer)
*/
@PublicEvolving
-public class TypeSerializerSchemaCompatibility<T, NS extends TypeSerializer<T>> {
+public class TypeSerializerSchemaCompatibility<T> {
/**
* Enum for the type of the compatibility.
@@ -77,7 +75,7 @@ public class TypeSerializerSchemaCompatibility<T, NS extends TypeSerializer<T>>
*
* @return a result that indicates migration is not required for the new serializer.
*/
- public static <T, NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> compatibleAsIs() {
+ public static <T> TypeSerializerSchemaCompatibility<T> compatibleAsIs() {
return new TypeSerializerSchemaCompatibility<>(Type.COMPATIBLE_AS_IS, null);
}
@@ -87,7 +85,7 @@ public class TypeSerializerSchemaCompatibility<T, NS extends TypeSerializer<T>>
*
* @return a result that indicates that the new serializer can be used after migrating the written bytes.
*/
- public static <T, NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> compatibleAfterMigration() {
+ public static <T> TypeSerializerSchemaCompatibility<T> compatibleAfterMigration() {
return new TypeSerializerSchemaCompatibility<>(Type.COMPATIBLE_AFTER_MIGRATION, null);
}
@@ -101,11 +99,11 @@ public class TypeSerializerSchemaCompatibility<T, NS extends TypeSerializer<T>>
*
* @return a result that indicates incompatibility between the new and previous serializer.
*/
- public static <T, NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> incompatible() {
+ public static <T> TypeSerializerSchemaCompatibility<T> incompatible() {
return new TypeSerializerSchemaCompatibility<>(Type.INCOMPATIBLE, null);
}
- private TypeSerializerSchemaCompatibility(Type resultType, @Nullable NS reconfiguredNewSerializer) {
+ private TypeSerializerSchemaCompatibility(Type resultType, @Nullable TypeSerializer<T> reconfiguredNewSerializer) {
this.resultType = Preconditions.checkNotNull(resultType);
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
index 4715d11..e004fa1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
@@ -126,11 +126,10 @@ public interface TypeSerializerSnapshot<T> {
* the format during the restore operation.
*
* @param newSerializer the new serializer to check.
- * @param <NS> the type of the new serializer
*
* @return the serializer compatibility result.
*/
- <NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> resolveSchemaCompatibility(NS newSerializer);
+ TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer);
// ------------------------------------------------------------------------
// read / write utilities
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerConfigSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerConfigSnapshot.java
index 6ce6aac..2774efe 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerConfigSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerConfigSnapshot.java
@@ -124,7 +124,7 @@ public final class GenericArraySerializerConfigSnapshot<C>
}
@Override
- protected TypeSerializerSchemaCompatibility<C[], GenericArraySerializer<C>>
+ protected TypeSerializerSchemaCompatibility<C[]>
outerCompatibility(GenericArraySerializer<C> serializer) {
return serializer.getComponentClass() == componentClass ?
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
index 6d576fd..7cda53f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
@@ -176,11 +176,11 @@ public final class ListSerializer<T> extends TypeSerializer<List<T>> {
}
// --------------------------------------------------------------------------------------------
- // Serializer configuration snapshotting & compatibility
+ // Serializer configuration snapshot & compatibility
// --------------------------------------------------------------------------------------------
@Override
- public CollectionSerializerConfigSnapshot<List<T>, T> snapshotConfiguration() {
+ public TypeSerializerSnapshot<List<T>> snapshotConfiguration() {
return new CollectionSerializerConfigSnapshot<>(elementSerializer);
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerConfigSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerConfigSnapshot.java
index ce3c449..d186953 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerConfigSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerConfigSnapshot.java
@@ -45,8 +45,8 @@ public final class EitherSerializerConfigSnapshot<L, R> extends CompositeTypeSer
@Override
@SuppressWarnings("unchecked")
- public <NS extends TypeSerializer<Either<L, R>>> TypeSerializerSchemaCompatibility<Either<L, R>, NS>
- resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<Either<L, R>> resolveSchemaCompatibility(
+ TypeSerializer<Either<L, R>> newSerializer) {
// this class was shared between the Java Either Serializer and the
// Scala Either serializer
@@ -61,7 +61,7 @@ public final class EitherSerializerConfigSnapshot<L, R> extends CompositeTypeSer
}
@SuppressWarnings("unchecked")
- private <NS extends TypeSerializer<Either<L, R>>> TypeSerializerSchemaCompatibility<Either<L, R>, NS> checkJavaSerializerCompatibility(
+ private TypeSerializerSchemaCompatibility<Either<L, R>> checkJavaSerializerCompatibility(
EitherSerializer<L, R> serializer) {
TypeSerializer<L> leftSerializer = serializer.getLeftSerializer();
@@ -70,8 +70,8 @@ public final class EitherSerializerConfigSnapshot<L, R> extends CompositeTypeSer
TypeSerializerSnapshot<L> leftSnapshot = (TypeSerializerSnapshot<L>) getNestedSerializersAndConfigs().get(0).f1;
TypeSerializerSnapshot<R> rightSnapshot = (TypeSerializerSnapshot<R>) getNestedSerializersAndConfigs().get(1).f1;
- TypeSerializerSchemaCompatibility<?, ?> leftCompatibility = leftSnapshot.resolveSchemaCompatibility(leftSerializer);
- TypeSerializerSchemaCompatibility<?, ?> rightCompatibility = rightSnapshot.resolveSchemaCompatibility(rightSerializer);
+ TypeSerializerSchemaCompatibility<?> leftCompatibility = leftSnapshot.resolveSchemaCompatibility(leftSerializer);
+ TypeSerializerSchemaCompatibility<?> rightCompatibility = rightSnapshot.resolveSchemaCompatibility(rightSerializer);
if (leftCompatibility.isCompatibleAsIs() && rightCompatibility.isCompatibleAsIs()) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerSnapshot.java
index b33408f..b0fece5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerSnapshot.java
@@ -103,7 +103,7 @@ public final class EitherSerializerSnapshot<L, R>
}
@Override
- protected TypeSerializerSchemaCompatibility<Either<L, R>, EitherSerializer<L, R>> outerCompatibility(EitherSerializer<L, R> serializer) {
+ protected TypeSerializerSchemaCompatibility<Either<L, R>> outerCompatibility(EitherSerializer<L, R> serializer) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
index c5713ac..a4908f9 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/SerializerTestBase.java
@@ -121,7 +121,7 @@ public abstract class SerializerTestBase<T> extends TestLogger {
new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader(), getSerializer());
}
- TypeSerializerSchemaCompatibility<T, ? extends TypeSerializer<T>> strategy = restoredConfig.resolveSchemaCompatibility(getSerializer());
+ TypeSerializerSchemaCompatibility<T> strategy = restoredConfig.resolveSchemaCompatibility(getSerializer());
assertTrue(strategy.isCompatibleAsIs());
TypeSerializer<T> restoreSerializer = restoredConfig.restoreSerializer();
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotMigrationTestBase.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotMigrationTestBase.java
index 40384ad..64474e2 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotMigrationTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotMigrationTestBase.java
@@ -79,7 +79,7 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
public void snapshotIsCompatibleWithTheCurrentSerializer() {
TypeSerializerSnapshot<ElementT> snapshot = snapshotUnderTest();
- TypeSerializerSchemaCompatibility<ElementT, TypeSerializer<ElementT>> result = snapshot.resolveSchemaCompatibility(testSpecification.createSerializer());
+ TypeSerializerSchemaCompatibility<ElementT> result = snapshot.resolveSchemaCompatibility(testSpecification.createSerializer());
assertTrue(result.isCompatibleAsIs());
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerTest.java
index 0fae269..03cc99e 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerTest.java
@@ -107,7 +107,7 @@ public class EnumSerializerTest extends TestLogger {
new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader(), serializer);
}
- TypeSerializerSchemaCompatibility<PublicEnum, ?> compatResult = restoredConfig.resolveSchemaCompatibility(serializer);
+ TypeSerializerSchemaCompatibility<PublicEnum> compatResult = restoredConfig.resolveSchemaCompatibility(serializer);
assertTrue(compatResult.isCompatibleAsIs());
assertEquals(PublicEnum.FOO.ordinal(), serializer.getValueToOrdinal().get(PublicEnum.FOO).intValue());
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
index ef421ef..ef66403 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
@@ -302,7 +302,7 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
}
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<SubTestUserClassA, ?> compatResult =
+ TypeSerializerSchemaCompatibility<SubTestUserClassA> compatResult =
pojoSerializerConfigSnapshot.resolveSchemaCompatibility(pojoSerializer2);
assertTrue(compatResult.isIncompatible());
}
@@ -345,7 +345,7 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
}
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestUserClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestUserClass> compatResult =
pojoSerializerConfigSnapshot.resolveSchemaCompatibility(pojoSerializer);
assertTrue(compatResult.isCompatibleAsIs());
@@ -392,7 +392,7 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
// reconfigure - check reconfiguration result and that subclass serializer cache is repopulated
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestUserClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestUserClass> compatResult =
pojoSerializerConfigSnapshot.resolveSchemaCompatibility(pojoSerializer);
assertTrue(compatResult.isCompatibleAsIs());
assertEquals(2, pojoSerializer.getSubclassSerializerCache().size());
@@ -457,7 +457,7 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
// 1) subclass serializer cache is repopulated
// 2) registrations also contain the now registered subclasses
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestUserClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestUserClass> compatResult =
pojoSerializerConfigSnapshot.resolveSchemaCompatibility(pojoSerializer);
assertTrue(compatResult.isCompatibleAsIs());
assertEquals(2, pojoSerializer.getSubclassSerializerCache().size());
@@ -535,7 +535,7 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
new HashMap<>()); // empty; irrelevant for this test
// reconfigure - check reconfiguration result and that fields are reordered to the previous order
- TypeSerializerSchemaCompatibility<TestUserClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestUserClass> compatResult =
mockPreviousConfigSnapshot.resolveSchemaCompatibility(pojoSerializer);
assertTrue(compatResult.isCompatibleAsIs());
int i = 0;
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerCompatibilityTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerCompatibilityTest.java
index 869941e..477843c 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerCompatibilityTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerCompatibilityTest.java
@@ -68,7 +68,7 @@ public class KryoSerializerCompatibilityTest {
}
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestClass> compatResult =
kryoSerializerConfigSnapshot.resolveSchemaCompatibility(kryoSerializerForA);
assertTrue(compatResult.isCompatibleAsIs());
}
@@ -114,7 +114,7 @@ public class KryoSerializerCompatibilityTest {
}
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestClassB, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestClassB> compatResult =
kryoSerializerConfigSnapshot.resolveSchemaCompatibility(kryoSerializerForB);
assertTrue(compatResult.isIncompatible());
}
@@ -277,7 +277,7 @@ public class KryoSerializerCompatibilityTest {
// reconfigure - check reconfiguration result and that registration id remains the same
@SuppressWarnings("unchecked")
- TypeSerializerSchemaCompatibility<TestClass, ?> compatResult =
+ TypeSerializerSchemaCompatibility<TestClass> compatResult =
kryoSerializerConfigSnapshot.resolveSchemaCompatibility(kryoSerializer);
assertTrue(compatResult.isCompatibleAsIs());
assertEquals(testClassId, kryoSerializer.getKryo().getRegistration(TestClass.class).getId());
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
index de9fcdc..a5f47d1 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
@@ -109,8 +109,8 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
}
@Override
- public <NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS>
- resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<T>
+ resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
if (!(newSerializer instanceof AvroSerializer)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
@@ -143,7 +143,7 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
* (@see <a href="https://avro.apache.org/docs/current/spec.html#Schema+Resolution">Schema Resolution</a>).
*/
@VisibleForTesting
- static <T, NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS> resolveSchemaCompatibility(
+ static <T> TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
Schema writerSchema,
Schema readerSchema) {
@@ -157,7 +157,7 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
return avroCompatibilityToFlinkCompatibility(compatibility);
}
- private static <T, NS extends TypeSerializer<T>> TypeSerializerSchemaCompatibility<T, NS>
+ private static <T> TypeSerializerSchemaCompatibility<T>
avroCompatibilityToFlinkCompatibility(SchemaPairCompatibility compatibility) {
switch (compatibility.getType()) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index ae4fbaa..4702919 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -232,7 +232,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
TypeSerializerSnapshot<K> keySerializerSnapshot = Preconditions.checkNotNull(
(TypeSerializerSnapshot<K>) metaInfoSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER));
- TypeSerializerSchemaCompatibility<K, ?> keyCompatibility =
+ TypeSerializerSchemaCompatibility<K> keyCompatibility =
keySerializerSnapshot.resolveSchemaCompatibility(broadcastStateKeySerializer);
if (keyCompatibility.isIncompatible()) {
throw new StateMigrationException("The new key serializer for broadcast state must not be incompatible.");
@@ -241,7 +241,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
TypeSerializerSnapshot<V> valueSerializerSnapshot = Preconditions.checkNotNull(
(TypeSerializerSnapshot<V>) metaInfoSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
- TypeSerializerSchemaCompatibility<V, ?> valueCompatibility =
+ TypeSerializerSchemaCompatibility<V> valueCompatibility =
valueSerializerSnapshot.resolveSchemaCompatibility(broadcastStateValueSerializer);
if (valueCompatibility.isIncompatible()) {
throw new StateMigrationException("The new value serializer for broadcast state must not be incompatible.");
@@ -601,7 +601,7 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
TypeSerializerSnapshot<S> stateSerializerSnapshot = Preconditions.checkNotNull(
(TypeSerializerSnapshot<S>) restoredSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
- TypeSerializerSchemaCompatibility<S, ?> stateCompatibility =
+ TypeSerializerSchemaCompatibility<S> stateCompatibility =
stateSerializerSnapshot.resolveSchemaCompatibility(newPartitionStateSerializer);
if (stateCompatibility.isIncompatible()) {
throw new StateMigrationException("The new state serializer for operator state must not be incompatible.");
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
index ebc79ac..4eff3a2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -209,7 +209,7 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
TypeSerializerSnapshot<T> serializerSnapshot = Preconditions.checkNotNull(
(TypeSerializerSnapshot<T>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(serializerKey));
- TypeSerializerSchemaCompatibility<T, ?> compatibilityResult =
+ TypeSerializerSchemaCompatibility<T> compatibilityResult =
serializerSnapshot.resolveSchemaCompatibility(byteOrderedElementSerializer);
if (compatibilityResult.isIncompatible()) {
@@ -281,7 +281,7 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
(TypeSerializerSnapshot<N>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString()));
- TypeSerializerSchemaCompatibility<N, ?> namespaceCompatibility =
+ TypeSerializerSchemaCompatibility<N> namespaceCompatibility =
namespaceSerializerSnapshot.resolveSchemaCompatibility(namespaceSerializer);
if (namespaceCompatibility.isIncompatible()) {
throw new StateMigrationException("For heap backends, the new namespace serializer must not be incompatible.");
@@ -294,7 +294,7 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
RegisteredKeyValueStateBackendMetaInfo.checkStateMetaInfo(restoredMetaInfoSnapshot, stateDesc);
- TypeSerializerSchemaCompatibility<V, ?> stateCompatibility =
+ TypeSerializerSchemaCompatibility<V> stateCompatibility =
stateSerializerSnapshot.resolveSchemaCompatibility(newStateSerializer);
if (stateCompatibility.isIncompatible()) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
index 74c55b6..de510de 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
@@ -936,7 +936,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
}
@Override
- public <NS extends TypeSerializer<TestType>> TypeSerializerSchemaCompatibility<TestType, NS> resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(TypeSerializer<TestType> newSerializer) {
switch (((TestSerializer) newSerializer).compatibilityType) {
case COMPATIBLE_AS_IS:
return TypeSerializerSchemaCompatibility.compatibleAsIs();
@@ -1048,7 +1048,8 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
}
@Override
- public <NS extends TypeSerializer<VoidNamespace>> TypeSerializerSchemaCompatibility<VoidNamespace, NS> resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<VoidNamespace>
+ resolveSchemaCompatibility(TypeSerializer<VoidNamespace> newSerializer) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
index f400dc3..83497c0 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
@@ -122,7 +122,7 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
}
def checkCompatibility(enumSourceA: String, enumSourceB: String)
- : TypeSerializerSchemaCompatibility[Enumeration#Value, _] = {
+ : TypeSerializerSchemaCompatibility[Enumeration#Value] = {
import EnumValueSerializerUpgradeTest._
val classLoader = compileAndLoadEnum(tempFolder.newFolder(), s"$enumName.scala", enumSourceA)
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 0063d31..50caa0d 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -1414,7 +1414,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
(TypeSerializerSnapshot<N>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString()));
- TypeSerializerSchemaCompatibility<N, ?> namespaceCompatibility =
+ TypeSerializerSchemaCompatibility<N> namespaceCompatibility =
namespaceSerializerSnapshot.resolveSchemaCompatibility(namespaceSerializer);
if (!namespaceCompatibility.isCompatibleAsIs()) {
throw new StateMigrationException("The new namespace serializer must be compatible.");
@@ -1427,7 +1427,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
RegisteredKeyValueStateBackendMetaInfo.checkStateMetaInfo(restoredMetaInfoSnapshot, stateDesc);
- TypeSerializerSchemaCompatibility<SV, ?> stateCompatibility =
+ TypeSerializerSchemaCompatibility<SV> stateCompatibility =
stateSerializerSnapshot.resolveSchemaCompatibility(stateSerializer);
if (stateCompatibility.isCompatibleAfterMigration()) {
@@ -1715,7 +1715,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
TypeSerializerSnapshot<T> serializerSnapshot = Preconditions.checkNotNull(
(TypeSerializerSnapshot<T>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(serializerKey));
- TypeSerializerSchemaCompatibility<T, ?> compatibilityResult =
+ TypeSerializerSchemaCompatibility<T> compatibilityResult =
serializerSnapshot.resolveSchemaCompatibility(byteOrderedElementSerializer);
// Since priority queue elements are written into RocksDB
diff --git a/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
index f813039..ac31872 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/migration/TypeSerializerSnapshotMigrationITCase.java
@@ -278,7 +278,7 @@ public class TypeSerializerSnapshotMigrationITCase extends SavepointMigrationTes
}
@Override
- public <NS extends TypeSerializer<Long>> TypeSerializerSchemaCompatibility<Long, NS> resolveSchemaCompatibility(NS newSerializer) {
+ public TypeSerializerSchemaCompatibility<Long> resolveSchemaCompatibility(TypeSerializer<Long> newSerializer) {
return (newSerializer instanceof TestSerializer)
? TypeSerializerSchemaCompatibility.compatibleAsIs()
: TypeSerializerSchemaCompatibility.incompatible();