You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ha...@apache.org on 2024/01/15 02:14:54 UTC
(flink) 07/32: [FLINK-30613][serializer] Migrate KryoSerializerSnapshot to implement new method of resolving schema compatibility
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1389856a6412eca4fd738e4cd37c6ef157e4dee1
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Tue Jan 24 01:00:48 2023 +0800
[FLINK-30613][serializer] Migrate KryoSerializerSnapshot to implement new method of resolving schema compatibility
---
.../typeutils/runtime/kryo/KryoSerializer.java | 16 ------------
.../runtime/kryo/KryoSerializerSnapshot.java | 30 ++++++++++------------
.../kryo/KryoSerializerCompatibilityTest.java | 8 ++++--
.../runtime/kryo/KryoSerializerSnapshotTest.java | 9 +++++--
4 files changed, 27 insertions(+), 36 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
index 3403f5e7469..3ca51a5c9f2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
@@ -250,22 +250,6 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
checkNotNull(kryoRegistrations, "Kryo registrations cannot be null.");
}
- Class<T> getType() {
- return type;
- }
-
- LinkedHashMap<Class<?>, SerializableSerializer<?>> getDefaultKryoSerializers() {
- return defaultSerializers;
- }
-
- LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> getDefaultKryoSerializerClasses() {
- return defaultSerializerClasses;
- }
-
- LinkedHashMap<String, KryoRegistration> getKryoRegistrations() {
- return kryoRegistrations;
- }
-
// ------------------------------------------------------------------------
@Override
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java
index 9696b1ef712..67de997c033 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java
@@ -33,11 +33,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.LinkedHashMap;
-import java.util.function.Function;
import static org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshotData.createFrom;
import static org.apache.flink.util.LinkedOptionalMap.mergeRightIntoLeft;
-import static org.apache.flink.util.LinkedOptionalMap.optionalMapOf;
/** {@link TypeSerializerSnapshot} for {@link KryoSerializer}. */
public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
@@ -92,24 +90,25 @@ public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer) {
- if (!(newSerializer instanceof KryoSerializer)) {
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof KryoSerializerSnapshot)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- KryoSerializer<T> kryoSerializer = (KryoSerializer<T>) newSerializer;
- if (kryoSerializer.getType() != snapshotData.getTypeClass()) {
+ KryoSerializerSnapshot<T> oldKryoSerializerSnapshot =
+ (KryoSerializerSnapshot<T>) oldSerializerSnapshot;
+ if (snapshotData.getTypeClass() != oldKryoSerializerSnapshot.snapshotData.getTypeClass()) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- return resolveSchemaCompatibility(kryoSerializer);
+ return resolveSchemaCompatibility(oldKryoSerializerSnapshot);
}
private TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- KryoSerializer<T> newSerializer) {
+ KryoSerializerSnapshot<T> oldKryoSerializerSnapshot) {
// merge the default serializers
final MergeResult<Class<?>, SerializableSerializer<?>> reconfiguredDefaultKryoSerializers =
mergeRightIntoLeft(
- snapshotData.getDefaultKryoSerializers(),
- optionalMapOf(newSerializer.getDefaultKryoSerializers(), Class::getName));
+ oldKryoSerializerSnapshot.snapshotData.getDefaultKryoSerializers(),
+ snapshotData.getDefaultKryoSerializers());
if (reconfiguredDefaultKryoSerializers.hasMissingKeys()) {
logMissingKeys(reconfiguredDefaultKryoSerializers);
@@ -120,10 +119,9 @@ public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
final MergeResult<Class<?>, Class<? extends Serializer<?>>>
reconfiguredDefaultKryoSerializerClasses =
mergeRightIntoLeft(
- snapshotData.getDefaultKryoSerializerClasses(),
- optionalMapOf(
- newSerializer.getDefaultKryoSerializerClasses(),
- Class::getName));
+ oldKryoSerializerSnapshot.snapshotData
+ .getDefaultKryoSerializerClasses(),
+ snapshotData.getDefaultKryoSerializerClasses());
if (reconfiguredDefaultKryoSerializerClasses.hasMissingKeys()) {
logMissingKeys(reconfiguredDefaultKryoSerializerClasses);
@@ -133,8 +131,8 @@ public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
// merge registration
final MergeResult<String, KryoRegistration> reconfiguredRegistrations =
mergeRightIntoLeft(
- snapshotData.getKryoRegistrations(),
- optionalMapOf(newSerializer.getKryoRegistrations(), Function.identity()));
+ oldKryoSerializerSnapshot.snapshotData.getKryoRegistrations(),
+ snapshotData.getKryoRegistrations());
if (reconfiguredRegistrations.hasMissingKeys()) {
logMissingKeys(reconfiguredRegistrations);
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 8cd3e09fa65..1468de6c1be 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
@@ -77,7 +77,9 @@ public class KryoSerializerCompatibilityTest {
@SuppressWarnings("unchecked")
TypeSerializerSchemaCompatibility<TestClassB> compatResult =
- kryoSerializerConfigSnapshot.resolveSchemaCompatibility(kryoSerializerForB);
+ kryoSerializerForB
+ .snapshotConfiguration()
+ .resolveSchemaCompatibility(kryoSerializerConfigSnapshot);
assertTrue(compatResult.isIncompatible());
}
@@ -246,7 +248,9 @@ public class KryoSerializerCompatibilityTest {
// reconfigure - check reconfiguration result and that registration id remains the same
@SuppressWarnings("unchecked")
TypeSerializerSchemaCompatibility<TestClass> compatResult =
- kryoSerializerConfigSnapshot.resolveSchemaCompatibility(kryoSerializer);
+ kryoSerializer
+ .snapshotConfiguration()
+ .resolveSchemaCompatibility(kryoSerializerConfigSnapshot);
assertTrue(compatResult.isCompatibleWithReconfiguredSerializer());
kryoSerializer = (KryoSerializer<TestClass>) compatResult.getReconfiguredSerializer();
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotTest.java
index c1761ce802e..94cd21a08d6 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotTest.java
@@ -106,7 +106,10 @@ public class KryoSerializerSnapshotTest {
new KryoSerializer<>(Animal.class, new ExecutionConfig());
assertThat(
- restoredSnapshot.resolveSchemaCompatibility(currentSerializer), isIncompatible());
+ currentSerializer
+ .snapshotConfiguration()
+ .resolveSchemaCompatibility(restoredSnapshot),
+ isIncompatible());
}
// -------------------------------------------------------------------------------------------------------
@@ -156,6 +159,8 @@ public class KryoSerializerSnapshotTest {
previousSerializer.snapshotConfiguration();
TypeSerializer<Animal> currentSerializer = new KryoSerializer<>(Animal.class, current);
- return previousSnapshot.resolveSchemaCompatibility(currentSerializer);
+ return currentSerializer
+ .snapshotConfiguration()
+ .resolveSchemaCompatibility(previousSnapshot);
}
}