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 2019/02/22 07:45:34 UTC
[flink] 03/17: [FLINK-11323] [core] Migrate KryoSerializer to use
new serialization compatibility abstractions
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 f436fbfe070dcea31be3560441b6f77c2ce54165
Author: Igal Shilman <ig...@data-artisans.com>
AuthorDate: Tue Jan 15 14:16:19 2019 +0100
[FLINK-11323] [core] Migrate KryoSerializer to use new serialization compatibility abstractions
This commit adds a new KryoSerializerSnapshot class and uses that
instead of the now deprecated KryoSerializerConfigSnapshot.
It also adds migration tests for the new KryoSerializerSnapshot,
extending the TypeSerializerSnapshotMigrationTestBase.
---
.../typeutils/runtime/kryo/KryoSerializer.java | 93 ++---
.../runtime/kryo/KryoSerializerSnapshot.java | 166 +++++++++
.../runtime/kryo/KryoSerializerSnapshotData.java | 377 +++++++++++++++++++++
.../TypeSerializerSnapshotMigrationTestBase.java | 6 +
.../runtime/kryo/KryoPojosForMigrationTests.java | 154 +++++++++
.../runtime/kryo/KryoSnapshotMigrationTest.java | 146 ++++++++
.../flink-1.6-kryo-type-serializer-custom-data | 1 +
.../flink-1.6-kryo-type-serializer-custom-snapshot | Bin 0 -> 4808 bytes
.../resources/flink-1.6-kryo-type-serializer-data | 1 +
...link-1.6-kryo-type-serializer-empty-config-data | Bin 0 -> 180 bytes
...-1.6-kryo-type-serializer-empty-config-snapshot | Bin 0 -> 2759 bytes
.../flink-1.6-kryo-type-serializer-snapshot | Bin 0 -> 3854 bytes
.../flink-1.7-kryo-type-serializer-custom-data | 1 +
.../flink-1.7-kryo-type-serializer-custom-snapshot | Bin 0 -> 4796 bytes
.../resources/flink-1.7-kryo-type-serializer-data | 1 +
...link-1.7-kryo-type-serializer-empty-config-data | Bin 0 -> 892 bytes
...-1.7-kryo-type-serializer-empty-config-snapshot | Bin 0 -> 2747 bytes
.../flink-1.7-kryo-type-serializer-snapshot | Bin 0 -> 3842 bytes
18 files changed, 906 insertions(+), 40 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 98b07e1..f1d3022 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
@@ -20,9 +20,10 @@ package org.apache.flink.api.java.typeutils.runtime.kryo;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.CompatibilityResult;
+import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.java.typeutils.AvroUtils;
import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
@@ -174,6 +175,36 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
}
}
+ // for KryoSerializerSnapshot
+ // ------------------------------------------------------------------------
+
+ KryoSerializer(Class<T> type,
+ LinkedHashMap<Class<?>, SerializableSerializer<?>> defaultSerializers,
+ LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> defaultSerializerClasses,
+ LinkedHashMap<String, KryoRegistration> kryoRegistrations) {
+
+ this.type = checkNotNull(type, "Type class cannot be null.");
+ this.defaultSerializerClasses = checkNotNull(defaultSerializerClasses, "Default serializer classes cannot be null.");
+ this.defaultSerializers = checkNotNull(defaultSerializers, "Default serializers cannot be null.");
+ this.kryoRegistrations = 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
@@ -452,46 +483,15 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
// --------------------------------------------------------------------------------------------
@Override
- public KryoSerializerConfigSnapshot<T> snapshotConfiguration() {
- return new KryoSerializerConfigSnapshot<>(type, kryoRegistrations);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot<?> configSnapshot) {
- if (configSnapshot instanceof KryoSerializerConfigSnapshot) {
- final KryoSerializerConfigSnapshot<T> config = (KryoSerializerConfigSnapshot<T>) configSnapshot;
-
- if (type.equals(config.getTypeClass())) {
- LinkedHashMap<String, KryoRegistration> reconfiguredRegistrations = config.getKryoRegistrations();
-
- // reconfigure by assuring that classes which were previously registered are registered
- // again in the exact same order; new class registrations will be appended.
- // this also overwrites any dummy placeholders that the restored old configuration has
- reconfiguredRegistrations.putAll(kryoRegistrations);
-
- // check if there is still any dummy placeholders even after reconfiguration;
- // if so, then this new Kryo serializer cannot read old data and is therefore incompatible
- for (Map.Entry<String, KryoRegistration> reconfiguredRegistrationEntry : reconfiguredRegistrations.entrySet()) {
- if (reconfiguredRegistrationEntry.getValue().isDummy()) {
- LOG.warn("The Kryo registration for a previously registered class {} does not have a " +
- "proper serializer, because its previous serializer cannot be loaded or is no " +
- "longer valid but a new serializer is not available", reconfiguredRegistrationEntry.getKey());
-
- return CompatibilityResult.requiresMigration();
- }
- }
-
- // there's actually no way to tell if new Kryo serializers are compatible with
- // the previous ones they overwrite; we can only signal compatibility and hope for the best
- this.kryoRegistrations = reconfiguredRegistrations;
- return CompatibilityResult.compatible();
- }
- }
-
- return CompatibilityResult.requiresMigration();
+ public TypeSerializerSnapshot<T> snapshotConfiguration() {
+ return new KryoSerializerSnapshot<>(
+ type,
+ defaultSerializers,
+ defaultSerializerClasses,
+ kryoRegistrations);
}
+ @Deprecated
public static final class KryoSerializerConfigSnapshot<T> extends KryoRegistrationSerializerConfigSnapshot<T> {
private static final int VERSION = 1;
@@ -510,6 +510,19 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
public int getVersion() {
return VERSION;
}
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ KryoSerializer<T> javaSerializedKryoSerializer = (KryoSerializer<T>) super.restoreSerializer();
+
+ KryoSerializerSnapshot<T> snapshot = new KryoSerializerSnapshot<>(
+ javaSerializedKryoSerializer.getType(),
+ javaSerializedKryoSerializer.getDefaultKryoSerializers(),
+ javaSerializedKryoSerializer.getDefaultKryoSerializerClasses(),
+ javaSerializedKryoSerializer.getKryoRegistrations());
+
+ return snapshot.resolveSchemaCompatibility(newSerializer);
+ }
}
// --------------------------------------------------------------------------------------------
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
new file mode 100644
index 0000000..37a0104
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshot.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime.kryo;
+
+import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.KryoRegistration;
+import org.apache.flink.api.java.typeutils.runtime.kryo.OptionalMap.MergeResult;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import com.esotericsoftware.kryo.Serializer;
+import org.slf4j.Logger;
+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.api.java.typeutils.runtime.kryo.OptionalMap.mergeRightIntoLeft;
+import static org.apache.flink.api.java.typeutils.runtime.kryo.OptionalMap.optionalMapOf;
+
+/**
+ * {@link TypeSerializerSnapshot} for {@link KryoSerializer}.
+ */
+public class KryoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KryoSerializerSnapshot.class);
+
+ private static final int VERSION = 2;
+
+ private KryoSerializerSnapshotData<T> snapshotData;
+
+ @SuppressWarnings("unused")
+ public KryoSerializerSnapshot() {
+ }
+
+ KryoSerializerSnapshot(Class<T> typeClass,
+ LinkedHashMap<Class<?>, SerializableSerializer<?>> defaultKryoSerializers,
+ LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses,
+ LinkedHashMap<String, KryoRegistration> kryoRegistrations) {
+
+ this.snapshotData = createFrom(typeClass, defaultKryoSerializers, defaultKryoSerializerClasses, kryoRegistrations);
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ snapshotData.writeSnapshotData(out);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ this.snapshotData = createFrom(in, userCodeClassLoader);
+ }
+
+ @Override
+ public TypeSerializer<T> restoreSerializer() {
+ return new KryoSerializer<>(
+ snapshotData.getTypeClass(),
+ snapshotData.getDefaultKryoSerializers().unwrapOptionals(),
+ snapshotData.getDefaultKryoSerializerClasses().unwrapOptionals(),
+ snapshotData.getKryoRegistrations().unwrapOptionals());
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ if (!(newSerializer instanceof KryoSerializer)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ KryoSerializer<T> kryoSerializer = (KryoSerializer<T>) newSerializer;
+ if (kryoSerializer.getType() != snapshotData.getTypeClass()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ return resolveSchemaCompatibility(kryoSerializer);
+ }
+
+ private TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(KryoSerializer<T> newSerializer) {
+ // merge the default serializers
+ final MergeResult<Class<?>, SerializableSerializer<?>> reconfiguredDefaultKryoSerializers = mergeRightIntoLeft(
+ snapshotData.getDefaultKryoSerializers(),
+ optionalMapOf(newSerializer.getDefaultKryoSerializers(), Class::getName));
+
+ if (reconfiguredDefaultKryoSerializers.hasMissingKeys()) {
+ logMissingKeys(reconfiguredDefaultKryoSerializers);
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ // merge default serializer classes
+ final MergeResult<Class<?>, Class<? extends Serializer<?>>> reconfiguredDefaultKryoSerializerClasses = mergeRightIntoLeft(
+ snapshotData.getDefaultKryoSerializerClasses(),
+ optionalMapOf(newSerializer.getDefaultKryoSerializerClasses(), Class::getName));
+
+ if (reconfiguredDefaultKryoSerializerClasses.hasMissingKeys()) {
+ logMissingKeys(reconfiguredDefaultKryoSerializerClasses);
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ // merge registration
+ final MergeResult<String, KryoRegistration> reconfiguredRegistrations = mergeRightIntoLeft(
+ snapshotData.getKryoRegistrations(),
+ optionalMapOf(newSerializer.getKryoRegistrations(), Function.identity()));
+
+ if (reconfiguredRegistrations.hasMissingKeys()) {
+ logMissingKeys(reconfiguredRegistrations);
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ // there are no missing keys, now we have to decide whether we are compatible as-is or we require reconfiguration.
+ return resolveSchemaCompatibility(
+ reconfiguredDefaultKryoSerializers,
+ reconfiguredDefaultKryoSerializerClasses,
+ reconfiguredRegistrations);
+ }
+
+ private TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ MergeResult<Class<?>, SerializableSerializer<?>> reconfiguredDefaultKryoSerializers,
+ MergeResult<Class<?>, Class<? extends Serializer<?>>> reconfiguredDefaultKryoSerializerClasses,
+ MergeResult<String, KryoRegistration> reconfiguredRegistrations) {
+
+ if (reconfiguredDefaultKryoSerializers.isOrderedSubset() &&
+ reconfiguredDefaultKryoSerializerClasses.isOrderedSubset() &&
+ reconfiguredRegistrations.isOrderedSubset()) {
+
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+
+ // reconfigure a new KryoSerializer
+ KryoSerializer<T> reconfiguredSerializer = new KryoSerializer<>(
+ snapshotData.getTypeClass(),
+ reconfiguredDefaultKryoSerializers.getMerged(),
+ reconfiguredDefaultKryoSerializerClasses.getMerged(),
+ reconfiguredRegistrations.getMerged());
+
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(reconfiguredSerializer);
+ }
+
+ private void logMissingKeys(MergeResult<?, ?> mergeResult) {
+ mergeResult.missingKeys().forEach(key -> LOG.warn("The Kryo registration for a previously registered class {} does not have a " +
+ "proper serializer, because its previous serializer cannot be loaded or is no " +
+ "longer valid but a new serializer is not available", key));
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotData.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotData.java
new file mode 100644
index 0000000..9587101
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializerSnapshotData.java
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime.kryo;
+
+import org.apache.flink.api.common.ExecutionConfig.SerializableSerializer;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.KryoRegistration;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.util.InstantiationUtil;
+
+import com.esotericsoftware.kryo.Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InvalidClassException;
+import java.util.LinkedHashMap;
+import java.util.Map.Entry;
+import java.util.function.Function;
+
+import static org.apache.flink.api.java.typeutils.runtime.kryo.OptionalMap.optionalMapOf;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+final class KryoSerializerSnapshotData<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KryoSerializerSnapshotData.class);
+
+ // --------------------------------------------------------------------------------------------
+ // Factories
+ // --------------------------------------------------------------------------------------------
+
+ static <T> KryoSerializerSnapshotData<T> createFrom(
+ Class<T> typeClass,
+ LinkedHashMap<Class<?>, SerializableSerializer<?>> defaultKryoSerializers,
+ LinkedHashMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses,
+ LinkedHashMap<String, KryoRegistration> kryoRegistrations) {
+
+ return new KryoSerializerSnapshotData<>(
+ typeClass,
+ optionalMapOf(defaultKryoSerializers, Class::getName),
+ optionalMapOf(defaultKryoSerializerClasses, Class::getName),
+ optionalMapOf(kryoRegistrations, Function.identity()));
+ }
+
+ static <T> KryoSerializerSnapshotData<T> createFrom(DataInputView in, ClassLoader cl) throws IOException {
+ Class<T> typeClass = readTypeClass(in, cl);
+ OptionalMap<String, KryoRegistration> kryoRegistrations = readKryoRegistrations(in, cl);
+ OptionalMap<Class<?>, SerializableSerializer<?>> defaultSerializer = readDefaultKryoSerializers(in, cl);
+ OptionalMap<Class<?>, Class<? extends Serializer<?>>> defaultSerializerClasses =
+ readDefaultKryoSerializerClasses(in, cl);
+
+ return new KryoSerializerSnapshotData<>(
+ typeClass,
+ defaultSerializer,
+ defaultSerializerClasses,
+ kryoRegistrations);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Fields
+ // --------------------------------------------------------------------------------------------
+
+ private final Class<T> typeClass;
+ private final OptionalMap<Class<?>, SerializableSerializer<?>> defaultKryoSerializers;
+ private final OptionalMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses;
+ private final OptionalMap<String, KryoRegistration> kryoRegistrations;
+
+ private KryoSerializerSnapshotData(Class<T> typeClass,
+ OptionalMap<Class<?>, SerializableSerializer<?>> defaultKryoSerializers,
+ OptionalMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses,
+ OptionalMap<String, KryoRegistration> kryoRegistrations) {
+
+ this.typeClass = typeClass;
+ this.defaultKryoSerializers = defaultKryoSerializers;
+ this.defaultKryoSerializerClasses = defaultKryoSerializerClasses;
+ this.kryoRegistrations = kryoRegistrations;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Getters
+ // --------------------------------------------------------------------------------------------
+
+ Class<T> getTypeClass() {
+ return typeClass;
+ }
+
+ OptionalMap<Class<?>, SerializableSerializer<?>> getDefaultKryoSerializers() {
+ return defaultKryoSerializers;
+ }
+
+ OptionalMap<Class<?>, Class<? extends Serializer<?>>> getDefaultKryoSerializerClasses() {
+ return defaultKryoSerializerClasses;
+ }
+
+ OptionalMap<String, KryoRegistration> getKryoRegistrations() {
+ return kryoRegistrations;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Write
+ // --------------------------------------------------------------------------------------------
+
+ void writeSnapshotData(DataOutputView out) throws IOException {
+ writeTypeClass(out);
+ writeKryoRegistrations(out, kryoRegistrations);
+ writeDefaultKryoSerializers(out, defaultKryoSerializers);
+ writeDefaultKryoSerializerClasses(out, defaultKryoSerializerClasses);
+ }
+
+ private void writeTypeClass(DataOutputView out) throws IOException {
+ out.writeUTF(typeClass.getName());
+ }
+
+ private static void writeKryoRegistrations(
+ DataOutputView out,
+ OptionalMap<String, KryoRegistration> kryoRegistrations) throws IOException {
+
+ out.writeInt(kryoRegistrations.size());
+ for (Entry<String, KryoRegistration> entry : kryoRegistrations.unwrapOptionals().entrySet()) {
+ out.writeUTF(entry.getKey());
+ KryoRegistrationUtil.writeKryoRegistration(entry.getValue(), out);
+ }
+ }
+
+ private void writeDefaultKryoSerializers(
+ DataOutputView out,
+ OptionalMap<Class<?>,
+ SerializableSerializer<?>> defaultKryoSerializers) throws IOException {
+
+ out.writeInt(defaultKryoSerializers.size());
+ for (Entry<Class<?>, SerializableSerializer<?>> entry : defaultKryoSerializers.unwrapOptionals().entrySet()) {
+ Class<?> javaClass = entry.getKey();
+ SerializableSerializer<?> serializerInstance = entry.getValue();
+
+ out.writeUTF(javaClass.getName());
+ try (final DataOutputViewStream outViewWrapper = new DataOutputViewStream(out)) {
+ InstantiationUtil.serializeObject(outViewWrapper, serializerInstance);
+ }
+ }
+ }
+
+ private static void writeDefaultKryoSerializerClasses(
+ DataOutputView out,
+ OptionalMap<Class<?>, Class<? extends Serializer<?>>> defaultKryoSerializerClasses)
+ throws IOException {
+
+ out.writeInt(defaultKryoSerializerClasses.size());
+
+ for (Entry<Class<?>, Class<? extends Serializer<?>>> entry : defaultKryoSerializerClasses.unwrapOptionals().entrySet()) {
+ Class<?> javaClass = entry.getKey();
+ Class<? extends Serializer<?>> serializerClass = entry.getValue();
+ out.writeUTF(javaClass.getName());
+ out.writeUTF(serializerClass.getName());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Read
+ // --------------------------------------------------------------------------------------------
+
+ private static <T> Class<T> readTypeClass(DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ return InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+ }
+
+ private static OptionalMap<String, KryoRegistration> readKryoRegistrations(
+ DataInputView in,
+ ClassLoader userCodeClassLoader) throws IOException {
+
+ OptionalMap<String, KryoRegistration> registrations = new OptionalMap<>();
+ final int size = in.readInt();
+ for (int i = 0; i < size; i++) {
+ final String name = in.readUTF();
+ KryoRegistration kryoRegistration = KryoRegistrationUtil.tryReadKryoRegistration(
+ in,
+ userCodeClassLoader);
+ registrations.put(name, name, kryoRegistration);
+ }
+
+ return registrations;
+ }
+
+ private static OptionalMap<Class<?>, SerializableSerializer<?>> readDefaultKryoSerializers(DataInputView in, ClassLoader cl) throws IOException {
+ OptionalMap<Class<?>, SerializableSerializer<?>> kryoSerializers = new OptionalMap<>();
+ final int size = in.readInt();
+ for (int i = 0; i < size; i++) {
+ final String className = in.readUTF();
+ Class<?> javaClass = null;
+ try {
+ javaClass = Class.forName(className, false, cl);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.warn("Cannot find registered class " + className + " for Kryo serialization in classpath.", e);
+ }
+ SerializableSerializer<?> kryoSerializer = null;
+ try {
+ try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
+ kryoSerializer = InstantiationUtil.deserializeObject(inViewWrapper, cl);
+ }
+ }
+ catch (Throwable e) {
+ LOG.warn("Cannot deserialize a previously serialized kryo serializer for the type " + className, e);
+ }
+ kryoSerializers.put(className, javaClass, kryoSerializer);
+ }
+ return kryoSerializers;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static OptionalMap<Class<?>, Class<? extends Serializer<?>>> readDefaultKryoSerializerClasses(
+ DataInputView in,
+ ClassLoader cl) throws IOException {
+
+ OptionalMap<Class<?>, Class<? extends Serializer<?>>> kryoSerializerClasses = new OptionalMap<>();
+ final int size = in.readInt();
+ for (int i = 0; i < size; i++) {
+ final String className = in.readUTF();
+ Class<?> typeClass = null;
+ try {
+ typeClass = Class.forName(className, false, cl);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.warn("Cannot find registered class " + className + " for Kryo serialization in classpath.", e);
+ }
+ final String kryoSerializerClassName = in.readUTF();
+ Class<? extends Serializer<?>> kryoSerializerClass = null;
+ try {
+ kryoSerializerClass = (Class<? extends Serializer<?>>) Class.forName(kryoSerializerClassName, false, cl);
+ }
+ catch (Throwable e) {
+ LOG.warn("Cannot find registered class " + className + " for Kryo serialization in classpath.", e);
+ }
+ kryoSerializerClasses.put(className, typeClass, kryoSerializerClass);
+ }
+ return kryoSerializerClasses;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helpers
+ // --------------------------------------------------------------------------------------------
+
+ private static final class KryoRegistrationUtil {
+
+ static void writeKryoRegistration(
+ KryoRegistration kryoRegistration,
+ DataOutputView out) throws IOException {
+
+ checkNotNull(kryoRegistration);
+ out.writeUTF(kryoRegistration.getRegisteredClass().getName());
+
+ final KryoRegistration.SerializerDefinitionType serializerDefinitionType =
+ kryoRegistration.getSerializerDefinitionType();
+
+ out.writeInt(serializerDefinitionType.ordinal());
+ switch (serializerDefinitionType) {
+ case UNSPECIFIED: {
+ // nothing else to write
+ break;
+ }
+ case CLASS: {
+ Class<? extends Serializer<?>> serializerClass = kryoRegistration.getSerializerClass();
+ assert serializerClass != null;
+ out.writeUTF(serializerClass.getName());
+ break;
+ }
+ case INSTANCE: {
+ try (final DataOutputViewStream outViewWrapper = new DataOutputViewStream(out)) {
+ InstantiationUtil.serializeObject(outViewWrapper, kryoRegistration.getSerializableSerializerInstance());
+ }
+ break;
+ }
+ default: {
+ throw new IllegalStateException(
+ "Unrecognized Kryo registration serializer definition type: " + serializerDefinitionType);
+ }
+ }
+ }
+
+ static KryoRegistration tryReadKryoRegistration(
+ DataInputView in,
+ ClassLoader userCodeClassLoader) throws IOException {
+
+ String registeredClassname = in.readUTF();
+ Class<?> registeredClass;
+ try {
+ registeredClass = Class.forName(registeredClassname, true, userCodeClassLoader);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.warn("Cannot find registered class " + registeredClassname + " for Kryo serialization in classpath;" +
+ " using a dummy class as a placeholder.", e);
+ return null;
+ }
+
+ final KryoRegistration.SerializerDefinitionType serializerDefinitionType =
+ KryoRegistration.SerializerDefinitionType.values()[in.readInt()];
+
+ switch (serializerDefinitionType) {
+ case UNSPECIFIED: {
+ return new KryoRegistration(registeredClass);
+ }
+ case CLASS: {
+ return tryReadWithSerializerClass(in, userCodeClassLoader, registeredClassname, registeredClass);
+ }
+ case INSTANCE: {
+ return tryReadWithSerializerInstance(in, userCodeClassLoader, registeredClassname, registeredClass);
+ }
+ default: {
+ throw new IllegalStateException(
+ "Unrecognized Kryo registration serializer definition type: " + serializerDefinitionType);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static KryoRegistration tryReadWithSerializerClass(
+ DataInputView in,
+ ClassLoader userCodeClassLoader,
+ String registeredClassname,
+ Class<?> registeredClass) throws IOException {
+ String serializerClassname = in.readUTF();
+ Class serializerClass;
+ try {
+ serializerClass = Class.forName(serializerClassname, true, userCodeClassLoader);
+ return new KryoRegistration(registeredClass, serializerClass);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.warn("Cannot find registered Kryo serializer class for class " + registeredClassname +
+ " in classpath; using a dummy Kryo serializer that should be replaced as soon as" +
+ " a new Kryo serializer for the class is present", e);
+ }
+ return null;
+ }
+
+ private static KryoRegistration tryReadWithSerializerInstance(
+ DataInputView in,
+ ClassLoader userCodeClassLoader,
+ String registeredClassname,
+ Class<?> registeredClass) throws IOException {
+ SerializableSerializer<? extends Serializer<?>> serializerInstance;
+
+ try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
+ serializerInstance = InstantiationUtil.deserializeObject(inViewWrapper, userCodeClassLoader);
+ return new KryoRegistration(registeredClass, serializerInstance);
+ }
+ catch (ClassNotFoundException e) {
+ LOG.warn("Cannot find registered Kryo serializer class for class " + registeredClassname +
+ " in classpath; using a dummy Kryo serializer that should be replaced as soon as" +
+ " a new Kryo serializer for the class is present", e);
+ }
+ catch (InvalidClassException e) {
+ LOG.warn("The registered Kryo serializer class for class " + registeredClassname +
+ " has changed and is no longer valid; using a dummy Kryo serializer that should be replaced" +
+ " as soon as a new Kryo serializer for the class is present.", e);
+
+ }
+ return null;
+ }
+
+ }
+}
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 31fc5ea..f52e411 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
@@ -271,6 +271,11 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
return this;
}
+ public TestSpecification<T> withTestDataCount(int expectedDataItmes) {
+ this.testDataCount = expectedDataItmes;
+ return this;
+ }
+
private TypeSerializer<T> createSerializer() {
try {
return (serializerProvider == null) ? serializerType.newInstance() : serializerProvider.get();
@@ -296,6 +301,7 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
public String toString() {
return String.format("%s , %s, %s", name, serializerType.getSimpleName(), snapshotClass.getSimpleName());
}
+
}
/**
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoPojosForMigrationTests.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoPojosForMigrationTests.java
new file mode 100644
index 0000000..c7b6cb0
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoPojosForMigrationTests.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime.kryo;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+import java.io.Serializable;
+
+/**
+ * POJOS needed for {@link KryoPojosForMigrationTests}.
+ */
+@SuppressWarnings("WeakerAccess")
+public class KryoPojosForMigrationTests {
+
+ public static abstract class Animal {
+ }
+
+ public static class Dog extends Animal {
+ private final String name;
+
+ public Dog(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+ }
+
+ public static class Cat extends Animal {
+ private final String name;
+ private final int age;
+
+ public Cat(String name, int age) {
+ this.name = name;
+ this.age = age;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public int getAge() {
+ return age;
+ }
+ }
+
+ public static class Parrot extends Animal {
+ private final String accent;
+
+ public Parrot(String accent) {
+ this.accent = accent;
+ }
+
+ public String getAccent() {
+ return accent;
+ }
+ }
+
+ // HousePets is registered explicitly in flink-1.6-kryo-type-serializer-*-* test resources.
+ @SuppressWarnings("unused")
+ public static class HousePets {
+ private final Dog dog;
+ private final Cat cat;
+
+ public HousePets(Dog dog, Cat cat) {
+ this.dog = dog;
+ this.cat = cat;
+ }
+
+ public Dog getDog() {
+ return dog;
+ }
+
+ public Cat getCat() {
+ return cat;
+ }
+ }
+
+ /**
+ * A Serializer that was registered during test data generation.
+ */
+ public static class ParrotKryoSerializer extends Serializer<Parrot> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void write(Kryo kryo, Output output, Parrot object) {
+ output.writeString(object.getAccent());
+ }
+
+ @Override
+ public Parrot read(Kryo kryo, Input input, Class<Parrot> type) {
+ return new Parrot(input.readString());
+ }
+ }
+
+ /**
+ * A Serializer that was registered during test data generation.
+ */
+ public static class DogKryoSerializer extends Serializer<Dog> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void write(Kryo kryo, Output output, Dog object) {
+ output.writeString(object.getName());
+ }
+
+ @Override
+ public Dog read(Kryo kryo, Input input, Class<Dog> type) {
+ return new Dog(input.readString());
+ }
+ }
+
+ /**
+ * A Serializer that is registered in migration tests.
+ */
+ public static class DogV2KryoSerializer extends Serializer<Dog> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void write(Kryo kryo, Output output, Dog object) {
+ output.writeString(object.getName());
+ }
+
+ @Override
+ public Dog read(Kryo kryo, Input input, Class<Dog> type) {
+ return new Dog(input.readString());
+ }
+ }
+
+
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSnapshotMigrationTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSnapshotMigrationTest.java
new file mode 100644
index 0000000..a982eca
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSnapshotMigrationTest.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime.kryo;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.Animal;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.Cat;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.Dog;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.DogV2KryoSerializer;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.Parrot;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoPojosForMigrationTests.ParrotKryoSerializer;
+import org.apache.flink.testutils.migration.MigrationVersion;
+
+import com.esotericsoftware.kryo.serializers.DefaultSerializers.StringSerializer;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility.compatibleAsIs;
+import static org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer;
+
+/**
+ * Tests migrations for {@link KryoSerializerSnapshot}.
+ */
+@SuppressWarnings("WeakerAccess")
+@RunWith(Parameterized.class)
+public class KryoSnapshotMigrationTest extends TypeSerializerSnapshotMigrationTestBase<KryoPojosForMigrationTests> {
+
+ public KryoSnapshotMigrationTest(TestSpecification<KryoPojosForMigrationTests> testSpecification) {
+ super(testSpecification);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Parameterized.Parameters(name = "Test Specification = {0}")
+ public static Collection<Object[]> testSpecifications() {
+
+ List<Object[]> specs = new ArrayList<>();
+
+ add(specs, "kryo-type-serializer-empty-config",
+ () -> new KryoSerializer<>(Animal.class, new ExecutionConfig()));
+
+ add(specs, "kryo-type-serializer-empty-config", () -> {
+
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.registerKryoType(DummyClassOne.class);
+ executionConfig.registerTypeWithKryoSerializer(DummyClassTwo.class, StringSerializer.class);
+
+ return new KryoSerializer<>(Animal.class, executionConfig);
+ }, COMPATIBLE_WITH_RECONFIGURED);
+
+ add(specs, "kryo-type-serializer", () -> {
+
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.registerKryoType(DummyClassOne.class);
+ executionConfig.registerKryoType(Dog.class);
+ executionConfig.registerKryoType(DummyClassTwo.class);
+ executionConfig.registerKryoType(Cat.class);
+ executionConfig.registerKryoType(Parrot.class);
+
+ return new KryoSerializer<>(Animal.class, executionConfig);
+ }, COMPATIBLE_WITH_RECONFIGURED);
+
+ add(specs, "kryo-type-serializer-custom", () -> {
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.registerKryoType(DummyClassOne.class);
+ executionConfig.registerTypeWithKryoSerializer(Dog.class, DogV2KryoSerializer.class);
+ executionConfig.registerKryoType(DummyClassTwo.class);
+ executionConfig.registerKryoType(Cat.class);
+ executionConfig.registerTypeWithKryoSerializer(Parrot.class, ParrotKryoSerializer.class);
+
+ return new KryoSerializer<>(Animal.class, executionConfig);
+ }, COMPATIBLE_WITH_RECONFIGURED);
+
+ return specs;
+ }
+
+ private static void add(List<Object[]> all, String name, Supplier<TypeSerializer<Animal>> supplier) {
+ add(all, name, supplier, compatibleAsIs());
+ }
+
+ private static void add(List<Object[]> all,
+ String name, Supplier<TypeSerializer<Animal>> supplier,
+ TypeSerializerSchemaCompatibility<Animal> expected) {
+
+ TestSpecification<Animal> flink16 = TestSpecification.<Animal>builder(
+ MigrationVersion.v1_6 + " " + name,
+ KryoSerializer.class,
+ KryoSerializerSnapshot.class,
+ MigrationVersion.v1_6)
+ .withNewSerializerProvider(supplier, expected)
+ .withSnapshotDataLocation("flink-1.6-" + name + "-snapshot")
+ .withTestData("flink-1.6-" + name + "-data", 2);
+
+ TestSpecification<Animal> flink17 = TestSpecification.<Animal>builder(
+ MigrationVersion.v1_7 + " " + name,
+ KryoSerializer.class,
+ KryoSerializerSnapshot.class,
+ MigrationVersion.v1_7)
+ .withNewSerializerProvider(supplier, expected)
+ .withSnapshotDataLocation("flink-1.7-" + name + "-snapshot")
+ .withTestData("flink-1.7-" + name + "-data", 2);
+
+ all.add(new Object[]{flink16});
+ all.add(new Object[]{flink17});
+ }
+
+ /**
+ * Dummy class to be registered in the tests.
+ */
+ public static final class DummyClassOne {
+
+ }
+
+ /**
+ * Dummy class to be registered in the tests.
+ */
+ public static final class DummyClassTwo {
+
+ }
+
+ private static final TypeSerializerSchemaCompatibility<Animal> COMPATIBLE_WITH_RECONFIGURED =
+ compatibleWithReconfiguredSerializer(new KryoSerializer<>(Animal.class, new ExecutionConfig()));
+}
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-data b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-data
new file mode 100644
index 0000000..b273ce4
--- /dev/null
+++ b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-data
@@ -0,0 +1 @@
+Scottis�Bobb�Scottis�Scottis�Scottis�Scottis�Scottis�Scottis�Scottis�Elvi�
\ No newline at end of file
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-snapshot b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-snapshot
new file mode 100644
index 0000000..de61118
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-custom-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-data b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-data
new file mode 100644
index 0000000..43cdb64
--- /dev/null
+++ b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-data
@@ -0,0 +1 @@
+PHome�Bob�
\ No newline at end of file
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-data b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-data
new file mode 100644
index 0000000..0c66b7d
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-data differ
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-snapshot b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-snapshot
new file mode 100644
index 0000000..eff068b
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-empty-config-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-snapshot b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-snapshot
new file mode 100644
index 0000000..2af53a7
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-kryo-type-serializer-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-data b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-data
new file mode 100644
index 0000000..b273ce4
--- /dev/null
+++ b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-data
@@ -0,0 +1 @@
+Scottis�Bobb�Scottis�Scottis�Scottis�Scottis�Scottis�Scottis�Scottis�Elvi�
\ No newline at end of file
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-snapshot b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-snapshot
new file mode 100644
index 0000000..4098c43
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-custom-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-data b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-data
new file mode 100644
index 0000000..d56bc15
--- /dev/null
+++ b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-data
@@ -0,0 +1 @@
+PHome�Bob�Bob�Bob�Bob�Bob�Bob�Bob�Bob�Bob�
\ No newline at end of file
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-data b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-data
new file mode 100644
index 0000000..62aa733
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-data differ
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-snapshot b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-snapshot
new file mode 100644
index 0000000..87f7ba9
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-empty-config-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-snapshot b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-snapshot
new file mode 100644
index 0000000..8117ac9
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-kryo-type-serializer-snapshot differ