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:41 UTC
[flink] 10/17: [FLINK-11485] [core] Refactor multiple nested
serializer compatibility resolution logic to CompositeTypeSerializerUtil
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 d5d45c8cc5924a6f387009eb6d0d053106055c3b
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Feb 19 14:08:03 2019 +0800
[FLINK-11485] [core] Refactor multiple nested serializer compatibility resolution logic to CompositeTypeSerializerUtil
This commit refactors the logic of resolving overall compatibility
results across multiple nested serializers of a composite serializer out
of the CompositeTypeSerializerSnapshot class.
This allows us to reuse this functionality when implementing the
compatibility check for the PojoSerializerSnapshot
---
.../typeutils/CompositeTypeSerializerSnapshot.java | 54 +---
.../typeutils/CompositeTypeSerializerUtil.java | 146 ++++++++++
.../typeutils/CompositeTypeSerializerUtilTest.java | 149 +++++++++++
.../SchemaCompatibilityTestingSerializer.java | 296 +++++++++++++++++++++
4 files changed, 597 insertions(+), 48 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java
index d14a80e..0e8b2f3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java
@@ -26,10 +26,10 @@ import org.apache.flink.api.common.typeutils.base.MapSerializer;
import org.apache.flink.api.java.typeutils.runtime.EitherSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.util.Preconditions;
import java.io.IOException;
+import static org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil.IntermediateCompatibilityResult;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -304,57 +304,15 @@ public abstract class CompositeTypeSerializerSnapshot<T, S extends TypeSerialize
TypeSerializer<?>[] newNestedSerializers,
TypeSerializerSnapshot<?>[] nestedSerializerSnapshots) {
- Preconditions.checkArgument(newNestedSerializers.length == nestedSerializerSnapshots.length,
- "Different number of new serializers and existing serializer snapshots.");
-
- TypeSerializer<?>[] reconfiguredNestedSerializers = new TypeSerializer[newNestedSerializers.length];
-
- // check nested serializers for compatibility
- boolean nestedSerializerRequiresMigration = false;
- boolean hasReconfiguredNestedSerializers = false;
- for (int i = 0; i < nestedSerializerSnapshots.length; i++) {
- TypeSerializerSchemaCompatibility<?> compatibility =
- resolveCompatibility(newNestedSerializers[i], nestedSerializerSnapshots[i]);
-
- // if any one of the new nested serializers is incompatible, we can just short circuit the result
- if (compatibility.isIncompatible()) {
- return TypeSerializerSchemaCompatibility.incompatible();
- }
-
- if (compatibility.isCompatibleAfterMigration()) {
- nestedSerializerRequiresMigration = true;
- } else if (compatibility.isCompatibleWithReconfiguredSerializer()) {
- hasReconfiguredNestedSerializers = true;
- reconfiguredNestedSerializers[i] = compatibility.getReconfiguredSerializer();
- } else if (compatibility.isCompatibleAsIs()) {
- reconfiguredNestedSerializers[i] = newNestedSerializers[i];
- } else {
- throw new IllegalStateException("Undefined compatibility type.");
- }
- }
-
- if (nestedSerializerRequiresMigration) {
- return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- }
+ IntermediateCompatibilityResult<T> intermediateResult =
+ CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(newNestedSerializers, nestedSerializerSnapshots);
- if (hasReconfiguredNestedSerializers) {
+ if (intermediateResult.isCompatibleWithReconfiguredSerializer()) {
@SuppressWarnings("unchecked")
- TypeSerializer<T> reconfiguredCompositeSerializer = createOuterSerializerWithNestedSerializers(reconfiguredNestedSerializers);
+ TypeSerializer<T> reconfiguredCompositeSerializer = createOuterSerializerWithNestedSerializers(intermediateResult.getNestedSerializers());
return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(reconfiguredCompositeSerializer);
}
- // ends up here if everything is compatible as is
- return TypeSerializerSchemaCompatibility.compatibleAsIs();
- }
-
- @SuppressWarnings("unchecked")
- private static <E> TypeSerializerSchemaCompatibility<E> resolveCompatibility(
- TypeSerializer<?> serializer,
- TypeSerializerSnapshot<?> snapshot) {
-
- TypeSerializer<E> typedSerializer = (TypeSerializer<E>) serializer;
- TypeSerializerSnapshot<E> typedSnapshot = (TypeSerializerSnapshot<E>) snapshot;
-
- return typedSnapshot.resolveSchemaCompatibility(typedSerializer);
+ return intermediateResult.getFinalResult();
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java
index f98bc72..8e1a65a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java
@@ -19,8 +19,11 @@
package org.apache.flink.api.common.typeutils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.Preconditions;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* Utilities for the {@link CompositeTypeSerializerSnapshot}.
@@ -62,4 +65,147 @@ public class CompositeTypeSerializerUtil {
NestedSerializersSnapshotDelegate delegate = new NestedSerializersSnapshotDelegate(nestedSnapshots);
compositeSnapshot.setNestedSerializersSnapshotDelegate(delegate);
}
+
+ /**
+ * Constructs an {@link IntermediateCompatibilityResult} with the given array of nested serializers and their
+ * corresponding serializer snapshots.
+ *
+ * <p>This result is considered "intermediate", because the actual final result is not yet built if it isn't
+ * defined. This is the case if the final result is supposed to be
+ * {@link TypeSerializerSchemaCompatibility#compatibleWithReconfiguredSerializer(TypeSerializer)}, where
+ * construction of the reconfigured serializer instance should be done by the caller.
+ *
+ * <p>For other cases, i.e. {@link TypeSerializerSchemaCompatibility#compatibleAsIs()},
+ * {@link TypeSerializerSchemaCompatibility#compatibleAfterMigration()}, and
+ * {@link TypeSerializerSchemaCompatibility#incompatible()}, these results are considered final.
+ *
+ * @param newNestedSerializers the new nested serializers to check for compatibility.
+ * @param nestedSerializerSnapshots the associated nested serializers' snapshots.
+ *
+ * @return the intermediate compatibility result of the new nested serializers.
+ */
+ public static <T> IntermediateCompatibilityResult<T> constructIntermediateCompatibilityResult(
+ TypeSerializer<?>[] newNestedSerializers,
+ TypeSerializerSnapshot<?>[] nestedSerializerSnapshots) {
+
+ Preconditions.checkArgument(newNestedSerializers.length == nestedSerializerSnapshots.length,
+ "Different number of new serializers and existing serializer snapshots.");
+
+ TypeSerializer<?>[] nestedSerializers = new TypeSerializer[newNestedSerializers.length];
+
+ // check nested serializers for compatibility
+ boolean nestedSerializerRequiresMigration = false;
+ boolean hasReconfiguredNestedSerializers = false;
+ for (int i = 0; i < nestedSerializerSnapshots.length; i++) {
+ TypeSerializerSchemaCompatibility<?> compatibility =
+ resolveCompatibility(newNestedSerializers[i], nestedSerializerSnapshots[i]);
+
+ // if any one of the new nested serializers is incompatible, we can just short circuit the result
+ if (compatibility.isIncompatible()) {
+ return IntermediateCompatibilityResult.definedIncompatibleResult();
+ }
+
+ if (compatibility.isCompatibleAfterMigration()) {
+ nestedSerializerRequiresMigration = true;
+ } else if (compatibility.isCompatibleWithReconfiguredSerializer()) {
+ hasReconfiguredNestedSerializers = true;
+ nestedSerializers[i] = compatibility.getReconfiguredSerializer();
+ } else if (compatibility.isCompatibleAsIs()) {
+ nestedSerializers[i] = newNestedSerializers[i];
+ } else {
+ throw new IllegalStateException("Undefined compatibility type.");
+ }
+ }
+
+ if (nestedSerializerRequiresMigration) {
+ return IntermediateCompatibilityResult.definedCompatibleAfterMigrationResult();
+ }
+
+ if (hasReconfiguredNestedSerializers) {
+ return IntermediateCompatibilityResult.undefinedReconfigureResult(nestedSerializers);
+ }
+
+ // ends up here if everything is compatible as is
+ return IntermediateCompatibilityResult.definedCompatibleAsIsResult(nestedSerializers);
+ }
+
+ public static class IntermediateCompatibilityResult<T> {
+
+ private final TypeSerializerSchemaCompatibility.Type compatibilityType;
+ private final TypeSerializer<?>[] nestedSerializers;
+
+ static <T> IntermediateCompatibilityResult<T> definedCompatibleAsIsResult(TypeSerializer<?>[] originalSerializers) {
+ return new IntermediateCompatibilityResult<>(TypeSerializerSchemaCompatibility.Type.COMPATIBLE_AS_IS, originalSerializers);
+ }
+
+ static <T> IntermediateCompatibilityResult<T> definedIncompatibleResult() {
+ return new IntermediateCompatibilityResult<>(TypeSerializerSchemaCompatibility.Type.INCOMPATIBLE, null);
+ }
+
+ static <T> IntermediateCompatibilityResult<T> definedCompatibleAfterMigrationResult() {
+ return new IntermediateCompatibilityResult<>(TypeSerializerSchemaCompatibility.Type.COMPATIBLE_AFTER_MIGRATION, null);
+ }
+
+ static <T> IntermediateCompatibilityResult<T> undefinedReconfigureResult(TypeSerializer<?>[] reconfiguredNestedSerializers) {
+ return new IntermediateCompatibilityResult<>(TypeSerializerSchemaCompatibility.Type.COMPATIBLE_WITH_RECONFIGURED_SERIALIZER, reconfiguredNestedSerializers);
+ }
+
+ private IntermediateCompatibilityResult(
+ TypeSerializerSchemaCompatibility.Type compatibilityType,
+ TypeSerializer<?>[] nestedSerializers) {
+ this.compatibilityType = checkNotNull(compatibilityType);
+ this.nestedSerializers = nestedSerializers;
+ }
+
+ public boolean isCompatibleWithReconfiguredSerializer() {
+ return compatibilityType == TypeSerializerSchemaCompatibility.Type.COMPATIBLE_WITH_RECONFIGURED_SERIALIZER;
+ }
+
+ public boolean isCompatibleAsIs() {
+ return compatibilityType == TypeSerializerSchemaCompatibility.Type.COMPATIBLE_AS_IS;
+ }
+
+ public boolean isCompatibleAfterMigration() {
+ return compatibilityType == TypeSerializerSchemaCompatibility.Type.COMPATIBLE_AFTER_MIGRATION;
+ }
+
+ public boolean isIncompatible() {
+ return compatibilityType == TypeSerializerSchemaCompatibility.Type.INCOMPATIBLE;
+ }
+
+ public TypeSerializerSchemaCompatibility<T> getFinalResult() {
+ checkState(
+ compatibilityType != TypeSerializerSchemaCompatibility.Type.COMPATIBLE_WITH_RECONFIGURED_SERIALIZER,
+ "unable to build final result if intermediate compatibility type is COMPATIBLE_WITH_RECONFIGURED_SERIALIZER.");
+ switch (compatibilityType) {
+ case COMPATIBLE_AS_IS:
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ case COMPATIBLE_AFTER_MIGRATION:
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ case INCOMPATIBLE:
+ return TypeSerializerSchemaCompatibility.incompatible();
+ default:
+ throw new IllegalStateException("unrecognized compatibility type.");
+ }
+ }
+
+ public TypeSerializer<?>[] getNestedSerializers() {
+ checkState(
+ compatibilityType == TypeSerializerSchemaCompatibility.Type.COMPATIBLE_AS_IS
+ || compatibilityType == TypeSerializerSchemaCompatibility.Type.COMPATIBLE_WITH_RECONFIGURED_SERIALIZER,
+ "only intermediate compatibility types COMPATIBLE_AS_IS and COMPATIBLE_WITH_RECONFIGURED_SERIALIZER have nested serializers.");
+ return nestedSerializers;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <E> TypeSerializerSchemaCompatibility<E> resolveCompatibility(
+ TypeSerializer<?> serializer,
+ TypeSerializerSnapshot<?> snapshot) {
+
+ TypeSerializer<E> typedSerializer = (TypeSerializer<E>) serializer;
+ TypeSerializerSnapshot<E> typedSnapshot = (TypeSerializerSnapshot<E>) snapshot;
+
+ return typedSnapshot.resolveSchemaCompatibility(typedSerializer);
+ }
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtilTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtilTest.java
new file mode 100644
index 0000000..37edcd7
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtilTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.common.typeutils;
+
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil.IntermediateCompatibilityResult;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.testutils.migration.SchemaCompatibilityTestingSerializer;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the {@link CompositeTypeSerializerUtil}.
+ */
+public class CompositeTypeSerializerUtilTest {
+
+ // ------------------------------------------------------------------------------------------------
+ // Tests for CompositeTypeSerializerUtil#constructIntermediateCompatibilityResult
+ // ------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testCompatibleAsIsIntermediateCompatibilityResult() {
+ final TypeSerializerSnapshot<?>[] testSerializerSnapshots = new TypeSerializerSnapshot<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ };
+
+ final TypeSerializer<?>[] testNewSerializers = new TypeSerializer<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ };
+
+ IntermediateCompatibilityResult<?> intermediateCompatibilityResult =
+ CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(testNewSerializers, testSerializerSnapshots);
+
+ assertTrue(intermediateCompatibilityResult.isCompatibleAsIs());
+ assertTrue(intermediateCompatibilityResult.getFinalResult().isCompatibleAsIs());
+ assertArrayEquals(testNewSerializers, intermediateCompatibilityResult.getNestedSerializers());
+ }
+
+ @Test
+ public void testCompatibleWithReconfiguredSerializerIntermediateCompatibilityResult() {
+ final TypeSerializerSnapshot<?>[] testSerializerSnapshots = new TypeSerializerSnapshot<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ };
+
+ final TypeSerializer<?>[] testNewSerializers = new TypeSerializer<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ new SchemaCompatibilityTestingSerializer.ReconfigurationRequiringSerializer<>(),
+ };
+
+ IntermediateCompatibilityResult<?> intermediateCompatibilityResult =
+ CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(testNewSerializers, testSerializerSnapshots);
+
+ assertTrue(intermediateCompatibilityResult.isCompatibleWithReconfiguredSerializer());
+
+ final TypeSerializer<?>[] expectedReconfiguredNestedSerializers = new TypeSerializer<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ };
+ assertArrayEquals(expectedReconfiguredNestedSerializers, intermediateCompatibilityResult.getNestedSerializers());
+ }
+
+ @Test
+ public void testCompatibleAfterMigrationIntermediateCompatibilityResult() {
+ final TypeSerializerSnapshot<?>[] testSerializerSnapshots = new TypeSerializerSnapshot<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ };
+
+ final TypeSerializer<?>[] testNewSerializers = new TypeSerializer<?>[] {
+ new SchemaCompatibilityTestingSerializer.ReconfigurationRequiringSerializer<>(),
+ new SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer<>(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer()
+ };
+
+ IntermediateCompatibilityResult<?> intermediateCompatibilityResult =
+ CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(testNewSerializers, testSerializerSnapshots);
+
+ assertTrue(intermediateCompatibilityResult.isCompatibleAfterMigration());
+ assertTrue(intermediateCompatibilityResult.getFinalResult().isCompatibleAfterMigration());
+ }
+
+ @Test
+ public void testIncompatibleIntermediateCompatibilityResult() {
+ final TypeSerializerSnapshot<?>[] testSerializerSnapshots = new TypeSerializerSnapshot<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ new SchemaCompatibilityTestingSerializer.InitialSerializer().snapshotConfiguration(),
+ };
+
+ final TypeSerializer<?>[] testNewSerializers = new TypeSerializer<?>[] {
+ new SchemaCompatibilityTestingSerializer.InitialSerializer(),
+ new SchemaCompatibilityTestingSerializer.IncompatibleSerializer<>(),
+ new SchemaCompatibilityTestingSerializer.ReconfigurationRequiringSerializer<>(),
+ new SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer()
+ };
+
+ IntermediateCompatibilityResult<?> intermediateCompatibilityResult =
+ CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(testNewSerializers, testSerializerSnapshots);
+
+ assertTrue(intermediateCompatibilityResult.isIncompatible());
+ assertTrue(intermediateCompatibilityResult.getFinalResult().isIncompatible());
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testGetFinalResultOnUndefinedReconfigureIntermediateCompatibilityResultFails() {
+ IntermediateCompatibilityResult<Integer> intermediateCompatibilityResult =
+ IntermediateCompatibilityResult.undefinedReconfigureResult(new TypeSerializer[]{ IntSerializer.INSTANCE });
+
+ intermediateCompatibilityResult.getFinalResult();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testGetNestedSerializersOnCompatibleAfterMigrationIntermediateCompatibilityResultFails() {
+ IntermediateCompatibilityResult<Integer> intermediateCompatibilityResult =
+ IntermediateCompatibilityResult.definedCompatibleAfterMigrationResult();
+
+ intermediateCompatibilityResult.getNestedSerializers();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testGetNestedSerializersOnIncompatibleIntermediateCompatibilityResultFails() {
+ IntermediateCompatibilityResult<Integer> intermediateCompatibilityResult =
+ IntermediateCompatibilityResult.definedIncompatibleResult();
+
+ intermediateCompatibilityResult.getNestedSerializers();
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/testutils/migration/SchemaCompatibilityTestingSerializer.java b/flink-core/src/test/java/org/apache/flink/testutils/migration/SchemaCompatibilityTestingSerializer.java
new file mode 100644
index 0000000..253eed7
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/testutils/migration/SchemaCompatibilityTestingSerializer.java
@@ -0,0 +1,296 @@
+/*
+ * 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.testutils.migration;
+
+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.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+
+/**
+ * The {@link SchemaCompatibilityTestingSerializer} is a mock serializer that can be used
+ * for schema compatibility and serializer upgrade related tests.
+ *
+ * <p>To use this, tests should always first instantiate an {@link InitialSerializer} serializer
+ * instance and then take a snapshot of it. Next, depending on what the new serializer is provided
+ * to the taken snapshot, tests can expect the following compatibility results:
+ * <ul>
+ * <li>{@link InitialSerializer} as the new serializer: the compatibility result will be
+ * {@link TypeSerializerSchemaCompatibility#compatibleAsIs()}.</li>
+ * <li>{@link UpgradedSchemaSerializer} as the new serializer: the compatibility result will be
+ * {@link TypeSerializerSchemaCompatibility#compatibleAfterMigration()}.</li>
+ * <li>{@link ReconfigurationRequiringSerializer} as the new serializer: the compatibility result
+ * will be {@link TypeSerializerSchemaCompatibility#compatibleWithReconfiguredSerializer(TypeSerializer)},
+ * with a new instance of the {@link InitialSerializer} as the wrapped reconfigured serializer.</li>
+ * <li>{@link IncompatibleSerializer} as the new serializer: the compatibility result will be
+ * {@link TypeSerializerSchemaCompatibility#incompatible()}.</li>
+ * </ul>
+ */
+public abstract class SchemaCompatibilityTestingSerializer<T> extends TypeSerializer<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ // ------------------------------------------------------------------------------------------------
+ // Irrelevant serializer methods
+ // ------------------------------------------------------------------------------------------------
+
+ @Override
+ public TypeSerializer<T> duplicate() {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+ }
+
+ @Override
+ public void serialize(T record, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public T deserialize(T reuse, DataInputView source) throws IOException {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public T deserialize(DataInputView source) throws IOException {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public T copy(T from) {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public T copy(T from, T reuse) {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public T createInstance() {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ @Override
+ public int getLength() {
+ throw new UnsupportedOperationException(
+ "this is a SchemaCompatibilityTestingSerializer; should have only been used for snapshotting / compatibility test purposes.");
+
+ }
+
+ // ------------------------------------------------------------------------------------------------
+ // Test serializer variants
+ // ------------------------------------------------------------------------------------------------
+
+ public static class InitialSerializer<T> extends SchemaCompatibilityTestingSerializer<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public TypeSerializerSnapshot<T> snapshotConfiguration() {
+ return new InitialSerializerSnapshot();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj != null && obj.getClass() == InitialSerializer.class;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
+
+ public static class UpgradedSchemaSerializer<T> extends SchemaCompatibilityTestingSerializer<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public TypeSerializerSnapshot<T> snapshotConfiguration() {
+ return new UpgradedSchemaSerializerSnapshot();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj != null && obj.getClass() == UpgradedSchemaSerializer.class;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
+
+ public static class ReconfigurationRequiringSerializer<T> extends SchemaCompatibilityTestingSerializer<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public TypeSerializerSnapshot<T> snapshotConfiguration() {
+ throw new UnsupportedOperationException(
+ "this is a ReconfigurationRequiringSerializer; should not have been snapshotted.");
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj != null && obj.getClass() == ReconfigurationRequiringSerializer.class;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
+
+ public static class IncompatibleSerializer<T> extends SchemaCompatibilityTestingSerializer<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public TypeSerializerSnapshot<T> snapshotConfiguration() {
+ throw new UnsupportedOperationException(
+ "this is a IncompatibleSerializer; should not have been snapshotted.");
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj != null && obj.getClass() == IncompatibleSerializer.class;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+ }
+
+ // ------------------------------------------------------------------------------------------------
+ // Test serializer snapshots
+ // ------------------------------------------------------------------------------------------------
+
+ public class InitialSerializerSnapshot implements TypeSerializerSnapshot<T> {
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {}
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {}
+
+ @Override
+ public TypeSerializer<T> restoreSerializer() {
+ return new SchemaCompatibilityTestingSerializer.InitialSerializer<>();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.InitialSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.ReconfigurationRequiringSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(new SchemaCompatibilityTestingSerializer.InitialSerializer<T>());
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.IncompatibleSerializer) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ // reaches here if newSerializer isn't a SchemaCompatibilityTestingSerializer
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ public class UpgradedSchemaSerializerSnapshot implements TypeSerializerSnapshot<T> {
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {}
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {}
+
+ @Override
+ public TypeSerializer<T> restoreSerializer() {
+ return new SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer<>();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.InitialSerializer) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.ReconfigurationRequiringSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(new SchemaCompatibilityTestingSerializer.UpgradedSchemaSerializer<T>());
+ }
+
+ if (newSerializer instanceof SchemaCompatibilityTestingSerializer.IncompatibleSerializer) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ // reaches here if newSerializer isn't a SchemaCompatibilityTestingSerializer
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+}