You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/02/21 08:53:35 UTC
[flink] branch master updated (397c9c6 -> 6996940)
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 397c9c6 [FLINK-11678] Let ExecutionGraphCache store only ArchivedExecutionGraphs
new 7c0989b [FLINK-11334][core] Migrate enum serializers to use new serialization compatibility abstractions
new 22f7aef [FLINK-11334] Remove old deserialization logic from new EnumSerializer Snapshot
new 2f6fbb5 [FLINK-11334] Fix compatibility check in EnumSerializer Snapshot and do proper reconfiguration
new c9f7282 [FLINK-11334] Fix EnumSerializer.equals()
new dd34f5e [FLINK-11334] Add good EnumSerializer.toString()
new ba5c495 [FLINK-11334] Add good TypeSerializerSchemaCompatibility.toString()
new 5095853 [FLINK-11334] Fix matcher in TypeSerializerSnapshotMigrationTestBase
new 96a0196 [FLINK-11334] Allow setting a compatibility matcher in TypeSerializerSnapshotMigrationTestBase
new 754cdae [FLINK-11334] Fix EnumSerializerSnapshotMigrationTest to check reconfiguration
new 8492bea [FLINK-11334] Initialize EnumSerializer in constructor
new 6996940 Fix checkstyle and warnings in EnumSerializer
The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../TypeSerializerSchemaCompatibility.java | 8 +
.../api/common/typeutils/base/EnumSerializer.java | 215 ++++++++++++++-------
.../TypeSerializerSnapshotMigrationTestBase.java | 51 ++++-
.../base/EnumSerializerSnapshotMigrationTest.java | 106 ++++++++++
.../common/typeutils/base/EnumSerializerTest.java | 28 ++-
.../typeutils/base/EnumSerializerUpgradeTest.java | 10 +-
.../flink/api/common/typeutils/base/TestEnum.java | 11 +-
.../test/resources/flink-1.6-enum-serializer-data | Bin 0 -> 40 bytes
.../resources/flink-1.6-enum-serializer-snapshot | Bin 0 -> 949 bytes
.../test/resources/flink-1.7-enum-serializer-data | Bin 0 -> 40 bytes
.../resources/flink-1.7-enum-serializer-snapshot | Bin 0 -> 937 bytes
11 files changed, 327 insertions(+), 102 deletions(-)
create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/util/event/EventListener.java => flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/TestEnum.java (82%)
create mode 100644 flink-core/src/test/resources/flink-1.6-enum-serializer-data
create mode 100644 flink-core/src/test/resources/flink-1.6-enum-serializer-snapshot
create mode 100644 flink-core/src/test/resources/flink-1.7-enum-serializer-data
create mode 100644 flink-core/src/test/resources/flink-1.7-enum-serializer-snapshot
[flink] 03/11: [FLINK-11334] Fix compatibility check in
EnumSerializer Snapshot and do proper reconfiguration
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2f6fbb5ad792a8ff7d0af5a368aa27480f18e402
Author: Igal Shilman <ig...@data-artisans.com>
AuthorDate: Wed Feb 20 09:46:06 2019 +0100
[FLINK-11334] Fix compatibility check in EnumSerializer Snapshot and do proper reconfiguration
This changes the new EnumSerializer Snapshot to hold Enums, instead of
their String representation (we to the mapping from and to Enum in the
read/write methods).
This also changes the compatibility checks to detect changes in enum
ordering and then report that only reconfiguration is required.
---
.../api/common/typeutils/base/EnumSerializer.java | 157 ++++++++++-----------
.../typeutils/base/EnumSerializerUpgradeTest.java | 10 +-
2 files changed, 77 insertions(+), 90 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index 6312780..e9818a5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -34,9 +34,12 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.Array;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Arrays;
+import java.util.EnumMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -72,11 +75,18 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
buildValues(enumClass.getEnumConstants());
}
+ EnumSerializer(Class<T> enumClass, T[] enumValues) {
+ this.enumClass = checkNotNull(enumClass);
+ checkArgument(Enum.class.isAssignableFrom(enumClass), "not an enum");
+
+ buildValues(enumValues);
+ }
+
private void buildValues(T[] values) {
this.values = values;
checkArgument(this.values.length > 0, "cannot use an empty enum");
- this.valueToOrdinal = new HashMap<>(values.length);
+ this.valueToOrdinal = new EnumMap<>(this.enumClass);
int i = 0;
for (T value : values) {
this.valueToOrdinal.put(value, i++);
@@ -140,7 +150,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@Override
public boolean equals(Object obj) {
- if(obj instanceof EnumSerializer) {
+ if (obj instanceof EnumSerializer) {
EnumSerializer<?> other = (EnumSerializer<?>) obj;
return other.enumClass == this.enumClass;
@@ -163,7 +173,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
if (this.values == null) {
this.values = enumClass.getEnumConstants();
- this.valueToOrdinal = new HashMap<>(values.length);
+ this.valueToOrdinal = new EnumMap<>(this.enumClass);
int i = 0;
for (T value : values) {
this.valueToOrdinal.put(value, i++);
@@ -183,22 +193,17 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
public static final class EnumSerializerSnapshot<T extends Enum<T>> implements TypeSerializerSnapshot<T> {
private static final int CURRENT_VERSION = 3;
- private List<String> previousEnumConstants;
+ private T[] previousEnums;
private Class<T> enumClass;
- @SuppressWarnings("WeakerAccess")
+ @SuppressWarnings("unused")
public EnumSerializerSnapshot() {
// this constructor is used when restoring from a checkpoint/savepoint.
}
- public EnumSerializerSnapshot(Class<T> enumClass, T[] enumConstantsArr) {
+ EnumSerializerSnapshot(Class<T> enumClass, T[] enums) {
this.enumClass = checkNotNull(enumClass);
- this.previousEnumConstants = buildEnumConstantsList(enumConstantsArr);
- }
-
- public EnumSerializerSnapshot(Class<T> enumClass, List<String> previousEnumConstants) {
- this.enumClass = checkNotNull(enumClass);
- this.previousEnumConstants = checkNotNull(previousEnumConstants);
+ this.previousEnums = checkNotNull(enums);
}
@Override
@@ -210,9 +215,9 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
public void writeSnapshot(DataOutputView out) throws IOException {
checkState(enumClass != null, "Enum class can not be null.");
out.writeUTF(enumClass.getName());
- out.writeInt(previousEnumConstants.size());
- for (String enumConstant : previousEnumConstants) {
- out.writeUTF(enumConstant);
+ out.writeInt(previousEnums.length);
+ for (T enumConstant : previousEnums) {
+ out.writeUTF(enumConstant.name());
}
}
@@ -221,93 +226,59 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
enumClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
int numEnumConstants = in.readInt();
- this.previousEnumConstants = new ArrayList<>(numEnumConstants);
+
+ @SuppressWarnings("unchecked")
+ T[] previousEnums = (T[]) Array.newInstance(enumClass, numEnumConstants);
for (int i = 0; i < numEnumConstants; i++) {
- previousEnumConstants.add(in.readUTF());
+ String enumName = in.readUTF();
+ try {
+ previousEnums[i] = Enum.valueOf(enumClass, enumName);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalStateException("Could not create a restore serializer for enum " +
+ enumClass +
+ ". Probably because an enum value was removed.");
+ }
}
+
+ this.previousEnums = previousEnums;
}
@Override
public TypeSerializer<T> restoreSerializer() {
checkState(enumClass != null, "Enum class can not be null.");
- return new EnumSerializer<>(enumClass);
+
+ return new EnumSerializer<>(enumClass, previousEnums);
}
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
- if (newSerializer instanceof EnumSerializer) {
- if (enumClass.equals(((EnumSerializer<T>) newSerializer).enumClass)) {
- EnumSerializer<T> newEnumSerializer = (EnumSerializer<T>) newSerializer;
-
- T[] reorderedEnumConstants = (T[]) Array.newInstance(enumClass, newEnumSerializer.values.length);
- Map<T, Integer> rebuiltEnumConstantToOrdinalMap = new HashMap<>(newEnumSerializer.values.length);
-
- if (previousEnumConstants.size() <= reorderedEnumConstants.length) {
- for (int i = 0; i < previousEnumConstants.size(); i++) {
- String previousEnumConstantStr = previousEnumConstants.get(i);
-
- try {
- // fetch the actual enum, and use it to populate the reconstructed bi-directional map
- T enumConstant = Enum.valueOf(enumClass, previousEnumConstantStr);
-
- reorderedEnumConstants[i] = enumConstant;
- rebuiltEnumConstantToOrdinalMap.put(enumConstant, i);
- } catch (IllegalArgumentException e) {
- // a previous enum constant no longer exists, and therefore requires migration
- return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- }
- }
- } else {
- return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- }
-
- // if there are new enum constants, append them to the end
- if (reorderedEnumConstants.length > previousEnumConstants.size()) {
- int appendedNewOrdinal = previousEnumConstants.size();
- for (T currentEnumConstant : newEnumSerializer.values) {
- if (!rebuiltEnumConstantToOrdinalMap.containsKey(currentEnumConstant)) {
- reorderedEnumConstants[appendedNewOrdinal] = currentEnumConstant;
- rebuiltEnumConstantToOrdinalMap.put(currentEnumConstant, appendedNewOrdinal);
- appendedNewOrdinal++;
- }
- }
- }
+ if (!(newSerializer instanceof EnumSerializer)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
- if (enumConstantsCompatible(reorderedEnumConstants, previousEnumConstants)) {
- return TypeSerializerSchemaCompatibility.compatibleAsIs();
- } else {
- EnumSerializer configuredSerializer = new EnumSerializer(enumClass);
- configuredSerializer.buildValues(reorderedEnumConstants);
- return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(configuredSerializer);
- }
- }
+ EnumSerializer<T> newEnumSerializer = (EnumSerializer<T>) newSerializer;
+ if (!enumClass.equals(newEnumSerializer.enumClass)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
}
- return TypeSerializerSchemaCompatibility.incompatible();
- }
+ T[] currentEnums = enumClass.getEnumConstants();
- private static <T extends Enum<T>> boolean enumConstantsCompatible(T[] reorderedEnumConstants, List<String> previousEnumConstants) {
- int idx = 0;
- if (reorderedEnumConstants.length != previousEnumConstants.size()) {
- return false;
- }
- for (String expected : previousEnumConstants) {
- String actual = reorderedEnumConstants[idx++].name();
- if (!actual.equals(expected)) {
- return false;
- }
+ if (Arrays.equals(previousEnums, currentEnums)) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
- return true;
- }
- private static <T extends Enum<T>> List<String> buildEnumConstantsList(T[] enumConstantsArr) {
- List<String> res = new ArrayList<>(enumConstantsArr.length);
+ Set<T> reconfiguredEnumSet = new LinkedHashSet<>(Arrays.asList(previousEnums));
+ reconfiguredEnumSet.addAll(Arrays.asList(currentEnums));
- for (T enumConstant : enumConstantsArr) {
- res.add(enumConstant.name());
- }
+ @SuppressWarnings("unchecked")
+ T[] reconfiguredEnums = reconfiguredEnumSet.toArray(
+ (T[]) Array.newInstance(enumClass, reconfiguredEnumSet.size()));
- return res;
+ EnumSerializer<T> reconfiguredSerializer = new EnumSerializer<>(
+ enumClass,
+ reconfiguredEnums);
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
+ reconfiguredSerializer);
}
}
@@ -375,7 +346,23 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
- return new EnumSerializerSnapshot<>(getTypeClass(), enumConstants).resolveSchemaCompatibility(newSerializer);
+
+ Class<T> enumClass = getTypeClass();
+
+ @SuppressWarnings("unchecked")
+ T[] previousEnums = (T[]) Array.newInstance(enumClass, enumConstants.size());
+
+ for (int i = 0; i < enumConstants.size(); i++) {
+ String enumName = enumConstants.get(i);
+ try {
+ previousEnums[i] = Enum.valueOf(enumClass, enumName);
+ } catch (IllegalArgumentException e) {
+ throw new RuntimeException("Could not create a restore serializer for enum " +
+ enumClass +
+ ". Probably because an enum value was removed.");
+ }
+ }
+ return new EnumSerializerSnapshot<>(enumClass, previousEnums).resolveSchemaCompatibility(newSerializer);
}
@Override
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
index f0d4014..93643b3 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
@@ -64,11 +64,11 @@ public class EnumSerializerUpgradeTest extends TestLogger {
}
/**
- * Check that removing enum fields requires migration
+ * Check that removing enum fields makes the snapshot incompatible
*/
- @Test
- public void checkRemovedField() throws Exception {
- Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_C).isCompatibleAfterMigration());
+ @Test(expected = IllegalStateException.class)
+ public void removingFieldShouldBeIncompatible() throws Exception {
+ Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_C).isIncompatible());
}
/**
@@ -76,7 +76,7 @@ public class EnumSerializerUpgradeTest extends TestLogger {
*/
@Test
public void checkDifferentFieldOrder() throws Exception {
- Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_D).isCompatibleAsIs());
+ Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_D).isCompatibleWithReconfiguredSerializer());
}
@SuppressWarnings("unchecked")
[flink] 05/11: [FLINK-11334] Add good EnumSerializer.toString()
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dd34f5e7fa558448d3ff5fad80e9f3fa410ecfc0
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 13:30:49 2019 +0100
[FLINK-11334] Add good EnumSerializer.toString()
This makes it easier to understand failing tests where serializers
are being compared.
---
.../apache/flink/api/common/typeutils/base/EnumSerializer.java | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index cfd342c..e12211a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -414,4 +414,12 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
Map<T, Integer> getValueToOrdinal() {
return valueToOrdinal;
}
+
+ @Override
+ public String toString() {
+ return "EnumSerializer{" +
+ "enumClass=" + enumClass +
+ ", values=" + Arrays.toString(values) +
+ '}';
+ }
}
[flink] 08/11: [FLINK-11334] Allow setting a compatibility matcher
in TypeSerializerSnapshotMigrationTestBase
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 96a0196449927338e6107d49c3329d88f91f020b
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 15:21:07 2019 +0100
[FLINK-11334] Allow setting a compatibility matcher in TypeSerializerSnapshotMigrationTestBase
---
.../TypeSerializerSnapshotMigrationTestBase.java | 45 ++++++++++++++++++++--
1 file changed, 42 insertions(+), 3 deletions(-)
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 55cf402..31fc5ea 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
TypeSerializerSchemaCompatibility<ElementT> result = snapshot.resolveSchemaCompatibility(testSpecification.createSerializer());
- assertThat(result, hasSameCompatibilityType(testSpecification.expectedCompatibilityResult));
+ assertThat(result, testSpecification.schemaCompatibilityMatcher);
}
@Test
@@ -203,7 +203,7 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
private final String name;
private final MigrationVersion testMigrationVersion;
private Supplier<? extends TypeSerializer<T>> serializerProvider;
- private TypeSerializerSchemaCompatibility<T> expectedCompatibilityResult;
+ private Matcher<TypeSerializerSchemaCompatibility<T>> schemaCompatibilityMatcher;
private String snapshotDataLocation;
private String testDataLocation;
private int testDataCount;
@@ -245,7 +245,13 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
Supplier<? extends TypeSerializer<T>> serializerProvider,
TypeSerializerSchemaCompatibility<T> expectedCompatibilityResult) {
this.serializerProvider = serializerProvider;
- this.expectedCompatibilityResult = expectedCompatibilityResult;
+ this.schemaCompatibilityMatcher = hasSameCompatibilityType(expectedCompatibilityResult);
+ return this;
+ }
+
+ public TestSpecification<T> withSchemaCompatibilityMatcher(
+ Matcher<TypeSerializerSchemaCompatibility<T>> schemaCompatibilityMatcher) {
+ this.schemaCompatibilityMatcher = schemaCompatibilityMatcher;
return this;
}
@@ -356,6 +362,39 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
* with the format "flink-<testVersion>-<specName>-<data/snapshot>",
* and each specification's test data count is assumed to always be 10.
*
+ * @param <T> type of the test data.
+ */
+ public <T> void addWithCompatibilityMatcher(
+ String name,
+ Class<? extends TypeSerializer> serializerClass,
+ Class<? extends TypeSerializerSnapshot> snapshotClass,
+ Supplier<? extends TypeSerializer<T>> serializerProvider,
+ Matcher<TypeSerializerSchemaCompatibility<T>> schemaCompatibilityMatcher) {
+ for (MigrationVersion testVersion : testVersions) {
+ testSpecifications.add(
+ TestSpecification.<T>builder(
+ getSpecNameForVersion(name, testVersion),
+ serializerClass,
+ snapshotClass,
+ testVersion)
+ .withNewSerializerProvider(serializerProvider)
+ .withSchemaCompatibilityMatcher(schemaCompatibilityMatcher)
+ .withSnapshotDataLocation(
+ String.format(DEFAULT_SNAPSHOT_FILENAME_FORMAT, testVersion, name))
+ .withTestData(
+ String.format(DEFAULT_TEST_DATA_FILENAME_FORMAT, testVersion, name),
+ DEFAULT_TEST_DATA_COUNT)
+ );
+ }
+ }
+
+ /**
+ * Adds a test specification to be tested for all specified test versions.
+ *
+ * <p>This method adds the specification with pre-defined snapshot and data filenames,
+ * with the format "flink-<testVersion>-<specName>-<data/snapshot>",
+ * and each specification's test data count is assumed to always be 10.
+ *
* @param name test specification name.
* @param serializerClass class of the current serializer.
* @param snapshotClass class of the current serializer snapshot class.
[flink] 01/11: [FLINK-11334][core] Migrate enum serializers to use
new serialization compatibility abstractions
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7c0989bf52ec23147a4b61d94fe878a11ab42cd0
Author: klion26 <qc...@gmail.com>
AuthorDate: Thu Feb 14 09:42:22 2019 +0800
[FLINK-11334][core] Migrate enum serializers to use new serialization compatibility abstractions
This commit touches EnumSerializer:
- replace an EnumSerializerSnapshot when calling snapshotConfiguration.
- add a test EnumSerializerSnapshotMigrationTest to varify the change.
- remove function EnumSerializer#ensureCompatibility.
---
.../api/common/typeutils/base/EnumSerializer.java | 208 ++++++++++++++++-----
.../base/EnumSerializerSnapshotMigrationTest.java | 55 ++++++
.../common/typeutils/base/EnumSerializerTest.java | 28 ++-
.../typeutils/base/EnumSerializerUpgradeTest.java | 4 +-
.../flink/api/common/typeutils/base/TestEnum.java | 24 +++
.../test/resources/flink-1.6-enum-serializer-data | Bin 0 -> 40 bytes
.../resources/flink-1.6-enum-serializer-snapshot | Bin 0 -> 949 bytes
.../test/resources/flink-1.7-enum-serializer-data | Bin 0 -> 40 bytes
.../resources/flink-1.7-enum-serializer-snapshot | Bin 0 -> 937 bytes
9 files changed, 250 insertions(+), 69 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index b8142a0..73aea19 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -18,28 +18,29 @@
package org.apache.flink.api.common.typeutils.base;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.lang.reflect.Array;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.typeutils.CompatibilityResult;
import org.apache.flink.api.common.typeutils.GenericTypeSerializerConfigSnapshot;
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.runtime.DataInputViewStream;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
@Internal
public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@@ -68,7 +69,11 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
this.enumClass = checkNotNull(enumClass);
checkArgument(Enum.class.isAssignableFrom(enumClass), "not an enum");
- this.values = enumClass.getEnumConstants();
+ buildValues(enumClass.getEnumConstants());
+ }
+
+ private void buildValues(T[] values) {
+ this.values = values;
checkArgument(this.values.length > 0, "cannot use an empty enum");
this.valueToOrdinal = new HashMap<>(values.length);
@@ -90,6 +95,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@Override
public T createInstance() {
+ checkState(values != null);
return values[0];
}
@@ -110,17 +116,20 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@Override
public void serialize(T record, DataOutputView target) throws IOException {
+ checkState(values != null);
// use our own maintained ordinals instead of the actual enum ordinal
target.writeInt(valueToOrdinal.get(record));
}
@Override
public T deserialize(DataInputView source) throws IOException {
+ checkState(values != null);
return values[source.readInt()];
}
@Override
public T deserialize(T reuse, DataInputView source) throws IOException {
+ checkState(values != null);
return values[source.readInt()];
}
@@ -167,64 +176,155 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
// --------------------------------------------------------------------------------------------
@Override
- public EnumSerializerConfigSnapshot<T> snapshotConfiguration() {
- return new EnumSerializerConfigSnapshot<>(enumClass, values);
+ public EnumSerializerSnapshot<T> snapshotConfiguration() {
+ return new EnumSerializerSnapshot<>(enumClass, values);
}
- @SuppressWarnings("unchecked")
- @Override
- public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot<?> configSnapshot) {
- if (configSnapshot instanceof EnumSerializerConfigSnapshot) {
- final EnumSerializerConfigSnapshot<T> config = (EnumSerializerConfigSnapshot<T>) configSnapshot;
+ public static final class EnumSerializerSnapshot<T extends Enum<T>> implements TypeSerializerSnapshot<T> {
+ private static final int CURRENT_VERSION = 3;
- if (enumClass.equals(config.getTypeClass())) {
+ private List<String> previousEnumConstants;
+ private Class<T> enumClass;
- T[] reorderedEnumConstants = (T[]) Array.newInstance(enumClass, this.values.length);
- Map<T, Integer> rebuiltEnumConstantToOrdinalMap = new HashMap<>(this.values.length);
+ @SuppressWarnings("WeakerAccess")
+ public EnumSerializerSnapshot() {
+ // this constructor is used when restoring from a checkpoint/savepoint.
+ }
- List<String> previousEnumConstants = config.getEnumConstants();
+ public EnumSerializerSnapshot(Class<T> enumClass, T[] enumConstantsArr) {
+ this.enumClass = checkNotNull(enumClass);
+ this.previousEnumConstants = buildEnumConstantsList(enumConstantsArr);
+ }
- if (previousEnumConstants.size() <= this.values.length) {
- for (int i = 0; i < previousEnumConstants.size(); i++) {
- String previousEnumConstantStr = previousEnumConstants.get(i);
+ public EnumSerializerSnapshot(Class<T> enumClass, List<String> previousEnumConstants) {
+ this.enumClass = checkNotNull(enumClass);
+ this.previousEnumConstants = checkNotNull(previousEnumConstants);
+ }
- try {
- // fetch the actual enum, and use it to populate the reconstructed bi-directional map
- T enumConstant = Enum.valueOf(enumClass, previousEnumConstantStr);
+ @Override
+ public int getCurrentVersion() {
+ return CURRENT_VERSION;
+ }
- reorderedEnumConstants[i] = enumConstant;
- rebuiltEnumConstantToOrdinalMap.put(enumConstant, i);
- } catch (IllegalArgumentException e) {
- // a previous enum constant no longer exists, and therefore requires migration
- return CompatibilityResult.requiresMigration();
- }
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ checkState(enumClass != null, "Enum class can not be null.");
+ out.writeUTF(enumClass.getName());
+ out.writeInt(previousEnumConstants.size());
+ for (String enumConstant : previousEnumConstants) {
+ out.writeUTF(enumConstant);
+ }
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ enumClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+
+ if (readVersion == 1) {
+ try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
+ try {
+ T[] legacyEnumConstants = InstantiationUtil.deserializeObject(inViewWrapper, userCodeClassLoader);
+ this.previousEnumConstants = buildEnumConstantsList(legacyEnumConstants);
+ } catch (ClassNotFoundException e) {
+ throw new IOException("The requested enum class cannot be found in classpath.", e);
+ } catch (IllegalArgumentException e) {
+ throw new IOException("A previously existing enum constant of "
+ + enumClass.getName() + " no longer exists.", e);
}
- } else {
- // some enum constants have been removed (because there are
- // fewer constants now), and therefore requires migration
- return CompatibilityResult.requiresMigration();
}
+ } else if (readVersion >= 2) {
+ int numEnumConstants = in.readInt();
+ this.previousEnumConstants = new ArrayList<>(numEnumConstants);
+ for (int i = 0; i < numEnumConstants; i++) {
+ previousEnumConstants.add(in.readUTF());
+ }
+ } else {
+ throw new IOException("Cannot deserialize EnumSerializerSnapshot with version " + readVersion);
+ }
+ }
- // if there are new enum constants, append them to the end
- if (this.values.length > previousEnumConstants.size()) {
- int appendedNewOrdinal = previousEnumConstants.size();
- for (T currentEnumConstant : this.values) {
- if (!rebuiltEnumConstantToOrdinalMap.containsKey(currentEnumConstant)) {
- reorderedEnumConstants[appendedNewOrdinal] = currentEnumConstant;
- rebuiltEnumConstantToOrdinalMap.put(currentEnumConstant, appendedNewOrdinal);
- appendedNewOrdinal++;
+ @Override
+ public TypeSerializer<T> restoreSerializer() {
+ checkState(enumClass != null, "Enum class can not be null.");
+ return new EnumSerializer<>(enumClass);
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ if (newSerializer instanceof EnumSerializer) {
+ if (enumClass.equals(((EnumSerializer<T>) newSerializer).enumClass)) {
+ EnumSerializer<T> newEnumSerializer = (EnumSerializer<T>) newSerializer;
+
+ T[] reorderedEnumConstants = (T[]) Array.newInstance(enumClass, newEnumSerializer.values.length);
+ Map<T, Integer> rebuiltEnumConstantToOrdinalMap = new HashMap<>(newEnumSerializer.values.length);
+
+ if (previousEnumConstants.size() <= reorderedEnumConstants.length) {
+ for (int i = 0; i < previousEnumConstants.size(); i++) {
+ String previousEnumConstantStr = previousEnumConstants.get(i);
+
+ try {
+ // fetch the actual enum, and use it to populate the reconstructed bi-directional map
+ T enumConstant = Enum.valueOf(enumClass, previousEnumConstantStr);
+
+ reorderedEnumConstants[i] = enumConstant;
+ rebuiltEnumConstantToOrdinalMap.put(enumConstant, i);
+ } catch (IllegalArgumentException e) {
+ // a previous enum constant no longer exists, and therefore requires migration
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ }
}
+ } else {
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ }
+
+ // if there are new enum constants, append them to the end
+ if (reorderedEnumConstants.length > previousEnumConstants.size()) {
+ int appendedNewOrdinal = previousEnumConstants.size();
+ for (T currentEnumConstant : newEnumSerializer.values) {
+ if (!rebuiltEnumConstantToOrdinalMap.containsKey(currentEnumConstant)) {
+ reorderedEnumConstants[appendedNewOrdinal] = currentEnumConstant;
+ rebuiltEnumConstantToOrdinalMap.put(currentEnumConstant, appendedNewOrdinal);
+ appendedNewOrdinal++;
+ }
+ }
+ }
+
+ if (enumConstantsCompatible(reorderedEnumConstants, previousEnumConstants)) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ } else {
+ EnumSerializer configuredSerializer = new EnumSerializer(enumClass);
+ configuredSerializer.buildValues(reorderedEnumConstants);
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(configuredSerializer);
}
}
+ }
+
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
- // if we reach here, we can simply reconfigure ourselves to be compatible
- this.values = reorderedEnumConstants;
- this.valueToOrdinal = rebuiltEnumConstantToOrdinalMap;
- return CompatibilityResult.compatible();
+ private static <T extends Enum<T>> boolean enumConstantsCompatible(T[] reorderedEnumConstants, List<String> previousEnumConstants) {
+ int idx = 0;
+ if (reorderedEnumConstants.length != previousEnumConstants.size()) {
+ return false;
+ }
+ for (String expected : previousEnumConstants) {
+ String actual = reorderedEnumConstants[idx++].name();
+ if (!actual.equals(expected)) {
+ return false;
+ }
}
+ return true;
}
- return CompatibilityResult.requiresMigration();
+ private static <T extends Enum<T>> List<String> buildEnumConstantsList(T[] enumConstantsArr) {
+ List<String> res = new ArrayList<>(enumConstantsArr.length);
+
+ for (T enumConstant : enumConstantsArr) {
+ res.add(enumConstant.name());
+ }
+
+ return res;
+ }
}
/**
@@ -235,6 +335,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
*
* @param <T> the enum type.
*/
+ @Deprecated
public static final class EnumSerializerConfigSnapshot<T extends Enum<T>>
extends GenericTypeSerializerConfigSnapshot<T> {
@@ -289,6 +390,11 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
}
@Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer) {
+ return new EnumSerializerSnapshot<>(getTypeClass(), enumConstants).resolveSchemaCompatibility(newSerializer);
+ }
+
+ @Override
public int getVersion() {
return VERSION;
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
new file mode 100644
index 0000000..f9e6955
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.base;
+
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
+import org.apache.flink.testutils.migration.MigrationVersion;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Collection;
+
+/**
+ * Migration tests for {@link EnumSerializer}.
+ */
+@RunWith(Parameterized.class)
+public class EnumSerializerSnapshotMigrationTest extends TypeSerializerSnapshotMigrationTestBase<TestEnum> {
+ private static final String SPEC_NAME = "enum-serializer";
+
+ public EnumSerializerSnapshotMigrationTest(TestSpecification<TestEnum> enumSerializer) {
+ super(enumSerializer);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Parameterized.Parameters(name = "Test Specification = {0}")
+ public static Collection<TestSpecification<?>> testSpecifications() {
+
+ final TestSpecifications testSpecifications = new TestSpecifications(MigrationVersion.v1_6, MigrationVersion.v1_7);
+
+ testSpecifications.add(
+ SPEC_NAME,
+ EnumSerializer.class,
+ EnumSerializer.EnumSerializerSnapshot.class,
+ () -> new EnumSerializer(TestEnum.class));
+
+ return testSpecifications.get();
+ }
+}
+
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 03cc99e..46e48d5 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
@@ -18,7 +18,6 @@
package org.apache.flink.api.common.typeutils.base;
-import org.apache.flink.api.common.typeutils.CompatibilityResult;
import org.apache.flink.api.common.typeutils.SerializerTestInstance;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil;
@@ -35,7 +34,6 @@ import java.io.ByteArrayOutputStream;
import java.util.Arrays;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class EnumSerializerTest extends TestLogger {
@@ -73,21 +71,22 @@ public class EnumSerializerTest extends TestLogger {
assertEquals(PublicEnum.PAULA.ordinal(), serializer.getValueToOrdinal().get(PublicEnum.PAULA).intValue());
// reconfigure and verify compatibility
- CompatibilityResult<PublicEnum> compatResult = serializer.ensureCompatibility(
- new EnumSerializer.EnumSerializerConfigSnapshot<>(PublicEnum.class, mockPreviousOrder));
- assertFalse(compatResult.isRequiresMigration());
+ EnumSerializer.EnumSerializerSnapshot serializerSnapshot = new EnumSerializer.EnumSerializerSnapshot(PublicEnum.class, mockPreviousOrder);
+ TypeSerializerSchemaCompatibility compatibility = serializerSnapshot.resolveSchemaCompatibility(serializer);
+ assertTrue(compatibility.isCompatibleWithReconfiguredSerializer());
// after reconfiguration, the order should be first the original BAR, PAULA, NATHANIEL,
// followed by the "new enum constants" FOO, PETER, EMMA
PublicEnum[] expectedOrder = {PublicEnum.BAR, PublicEnum.PAULA, PublicEnum.NATHANIEL, PublicEnum.FOO, PublicEnum.PETER, PublicEnum.EMMA};
+ EnumSerializer<PublicEnum> configuredSerializer = (EnumSerializer<PublicEnum>) compatibility.getReconfiguredSerializer();
int i = 0;
for (PublicEnum constant : expectedOrder) {
- assertEquals(i, serializer.getValueToOrdinal().get(constant).intValue());
+ assertEquals(i, configuredSerializer.getValueToOrdinal().get(constant).intValue());
i++;
}
- assertTrue(Arrays.equals(expectedOrder, serializer.getValues()));
+ assertTrue(Arrays.equals(expectedOrder, configuredSerializer.getValues()));
}
@Test
@@ -162,24 +161,21 @@ public class EnumSerializerTest extends TestLogger {
assertEquals(PublicEnum.PAULA.ordinal(), serializer.getValueToOrdinal().get(PublicEnum.PAULA).intValue());
// reconfigure and verify compatibility
- CompatibilityResult<PublicEnum> compatResult = serializer.ensureCompatibility(
- new EnumSerializer.EnumSerializerConfigSnapshot<>(PublicEnum.class, mockPreviousOrder));
- assertFalse(compatResult.isRequiresMigration());
-
- // serialize and deserialize again the serializer
- byte[] serializedSerializer = InstantiationUtil.serializeObject(serializer);
- serializer = InstantiationUtil.deserializeObject(serializedSerializer, Thread.currentThread().getContextClassLoader());
+ EnumSerializer.EnumSerializerSnapshot serializerSnapshot = new EnumSerializer.EnumSerializerSnapshot(PublicEnum.class, mockPreviousOrder);
+ TypeSerializerSchemaCompatibility compatibility = serializerSnapshot.resolveSchemaCompatibility(serializer);
+ assertTrue(compatibility.isCompatibleWithReconfiguredSerializer());
// verify that after the serializer was read, the reconfigured constant ordering is untouched
PublicEnum[] expectedOrder = {PublicEnum.BAR, PublicEnum.PAULA, PublicEnum.NATHANIEL, PublicEnum.FOO, PublicEnum.PETER, PublicEnum.EMMA};
+ EnumSerializer<PublicEnum> configuredSerializer = (EnumSerializer<PublicEnum>) compatibility.getReconfiguredSerializer();
int i = 0;
for (PublicEnum constant : expectedOrder) {
- assertEquals(i, serializer.getValueToOrdinal().get(constant).intValue());
+ assertEquals(i, configuredSerializer.getValueToOrdinal().get(constant).intValue());
i++;
}
- assertTrue(Arrays.equals(expectedOrder, serializer.getValues()));
+ assertTrue(Arrays.equals(expectedOrder, configuredSerializer.getValues()));
}
@SafeVarargs
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
index fb11945..f0d4014 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
@@ -60,7 +60,7 @@ public class EnumSerializerUpgradeTest extends TestLogger {
*/
@Test
public void checkAppendedField() throws Exception {
- Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_B).isCompatibleAsIs());
+ Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_B).isCompatibleWithReconfiguredSerializer());
}
/**
@@ -68,7 +68,7 @@ public class EnumSerializerUpgradeTest extends TestLogger {
*/
@Test
public void checkRemovedField() throws Exception {
- Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_C).isIncompatible());
+ Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_C).isCompatibleAfterMigration());
}
/**
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/TestEnum.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/TestEnum.java
new file mode 100644
index 0000000..b93cf2b
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/TestEnum.java
@@ -0,0 +1,24 @@
+/*
+ * 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.base;
+
+public enum TestEnum {
+ FOO, BAR, PETER, PAULA, NATHANIEL, EMMA
+}
+
diff --git a/flink-core/src/test/resources/flink-1.6-enum-serializer-data b/flink-core/src/test/resources/flink-1.6-enum-serializer-data
new file mode 100644
index 0000000..60d7723
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-enum-serializer-data differ
diff --git a/flink-core/src/test/resources/flink-1.6-enum-serializer-snapshot b/flink-core/src/test/resources/flink-1.6-enum-serializer-snapshot
new file mode 100644
index 0000000..9639d72
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-enum-serializer-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.7-enum-serializer-data b/flink-core/src/test/resources/flink-1.7-enum-serializer-data
new file mode 100644
index 0000000..97a46a4
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-enum-serializer-data differ
diff --git a/flink-core/src/test/resources/flink-1.7-enum-serializer-snapshot b/flink-core/src/test/resources/flink-1.7-enum-serializer-snapshot
new file mode 100644
index 0000000..b8add7c1
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-enum-serializer-snapshot differ
[flink] 10/11: [FLINK-11334] Initialize EnumSerializer in
constructor
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8492bea585a89385b8f9069ff825ed2de940c68a
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 15:22:37 2019 +0100
[FLINK-11334] Initialize EnumSerializer in constructor
We also don't check the state on every call anymore.
---
.../flink/api/common/typeutils/base/EnumSerializer.java | 14 ++------------
1 file changed, 2 insertions(+), 12 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index e12211a..823d4bf 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -69,21 +69,14 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
private T[] values;
public EnumSerializer(Class<T> enumClass) {
- this.enumClass = checkNotNull(enumClass);
- checkArgument(Enum.class.isAssignableFrom(enumClass), "not an enum");
-
- buildValues(enumClass.getEnumConstants());
+ this(enumClass, enumClass.getEnumConstants());
}
EnumSerializer(Class<T> enumClass, T[] enumValues) {
this.enumClass = checkNotNull(enumClass);
+ this.values = checkNotNull(enumValues);
checkArgument(Enum.class.isAssignableFrom(enumClass), "not an enum");
- buildValues(enumValues);
- }
-
- private void buildValues(T[] values) {
- this.values = values;
checkArgument(this.values.length > 0, "cannot use an empty enum");
this.valueToOrdinal = new EnumMap<>(this.enumClass);
@@ -126,20 +119,17 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@Override
public void serialize(T record, DataOutputView target) throws IOException {
- checkState(values != null);
// use our own maintained ordinals instead of the actual enum ordinal
target.writeInt(valueToOrdinal.get(record));
}
@Override
public T deserialize(DataInputView source) throws IOException {
- checkState(values != null);
return values[source.readInt()];
}
@Override
public T deserialize(T reuse, DataInputView source) throws IOException {
- checkState(values != null);
return values[source.readInt()];
}
[flink] 11/11: Fix checkstyle and warnings in EnumSerializer
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 69969409857f11e97cb490d083ab54cc24596b7c
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 15:26:06 2019 +0100
Fix checkstyle and warnings in EnumSerializer
---
.../api/common/typeutils/base/EnumSerializer.java | 21 +++++++++++----------
1 file changed, 11 insertions(+), 10 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index 823d4bf..6b188b0 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -28,7 +28,6 @@ import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.util.InstantiationUtil;
-import org.apache.flink.util.Preconditions;
import java.io.IOException;
import java.io.ObjectInputStream;
@@ -45,6 +44,9 @@ import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
+/**
+ * {@link TypeSerializer} for Java enums.
+ */
@Internal
public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@@ -53,7 +55,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
private final Class<T> enumClass;
/**
- * Maintain our own map of enum value to their ordinal, instead of directly using {@link Enum#ordinal}.
+ * Maintain our own map of enum value to their ordinal, instead of directly using {@link Enum#ordinal()}.
* This allows us to maintain backwards compatibility for previous serialized data in the case that the
* order of enum constants was changed or new constants were added.
*
@@ -72,7 +74,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
this(enumClass, enumClass.getEnumConstants());
}
- EnumSerializer(Class<T> enumClass, T[] enumValues) {
+ private EnumSerializer(Class<T> enumClass, T[] enumValues) {
this.enumClass = checkNotNull(enumClass);
this.values = checkNotNull(enumValues);
checkArgument(Enum.class.isAssignableFrom(enumClass), "not an enum");
@@ -181,6 +183,9 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
return new EnumSerializerSnapshot<>(enumClass, values);
}
+ /**
+ * {@link TypeSerializerSnapshot} for {@link EnumSerializer}.
+ */
public static final class EnumSerializerSnapshot<T extends Enum<T>> implements TypeSerializerSnapshot<T> {
private static final int CURRENT_VERSION = 3;
@@ -276,7 +281,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
/**
* Configuration snapshot of a serializer for enumerations.
*
- * Configuration contains the enum class, and an array of the enum's constants
+ * <p>Configuration contains the enum class, and an array of the enum's constants
* that existed when the configuration snapshot was taken.
*
* @param <T> the enum type.
@@ -290,13 +295,9 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
private List<String> enumConstants;
/** This empty nullary constructor is required for deserializing the configuration. */
+ @SuppressWarnings("unused")
public EnumSerializerConfigSnapshot() {}
- public EnumSerializerConfigSnapshot(Class<T> enumClass, T[] enumConstantsArr) {
- super(enumClass);
- this.enumConstants = buildEnumConstantsList(Preconditions.checkNotNull(enumConstantsArr));
- }
-
@Override
public void write(DataOutputView out) throws IOException {
super.write(out);
@@ -366,7 +367,7 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
return new int[] {VERSION, 1};
}
- public List<String> getEnumConstants() {
+ List<String> getEnumConstants() {
return enumConstants;
}
[flink] 06/11: [FLINK-11334] Add good
TypeSerializerSchemaCompatibility.toString()
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ba5c495ce02da7738e4008fdcefacd1194b871fd
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 13:31:38 2019 +0100
[FLINK-11334] Add good TypeSerializerSchemaCompatibility.toString()
This makes it easier to read messages of failing tests.
---
.../api/common/typeutils/TypeSerializerSchemaCompatibility.java | 8 ++++++++
1 file changed, 8 insertions(+)
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 4bb4aa0..aac156e 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
@@ -177,4 +177,12 @@ public class TypeSerializerSchemaCompatibility<T> {
public boolean isIncompatible() {
return resultType == Type.INCOMPATIBLE;
}
+
+ @Override
+ public String toString() {
+ return "TypeSerializerSchemaCompatibility{" +
+ "resultType=" + resultType +
+ ", reconfiguredNewSerializer=" + reconfiguredNewSerializer +
+ '}';
+ }
}
[flink] 04/11: [FLINK-11334] Fix EnumSerializer.equals()
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c9f7282219221bf8fe920efb49766a85ce5a0ca8
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 13:30:10 2019 +0100
[FLINK-11334] Fix EnumSerializer.equals()
Before, it wasn't comparing the ordering of enum values, i.e. the
mapping of indices to enum values.
---
.../org/apache/flink/api/common/typeutils/base/EnumSerializer.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index e9818a5..cfd342c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -153,7 +153,8 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
if (obj instanceof EnumSerializer) {
EnumSerializer<?> other = (EnumSerializer<?>) obj;
- return other.enumClass == this.enumClass;
+ return other.enumClass == this.enumClass &&
+ Arrays.equals(values, other.values);
} else {
return false;
}
[flink] 07/11: [FLINK-11334] Fix matcher in
TypeSerializerSnapshotMigrationTestBase
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5095853a50424be5cfe5f2253e07c6e32dcf0ff0
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 15:20:35 2019 +0100
[FLINK-11334] Fix matcher in TypeSerializerSnapshotMigrationTestBase
---
.../common/typeutils/TypeSerializerSnapshotMigrationTestBase.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
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 0548d2f..55cf402 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
@@ -443,7 +443,7 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
// Utilities
// --------------------------------------------------------------------------------------------------------------
- private <T> Matcher<TypeSerializerSchemaCompatibility<T>> hasSameCompatibilityType(TypeSerializerSchemaCompatibility<T> expectedCompatibilty) {
+ private static <T> Matcher<TypeSerializerSchemaCompatibility<T>> hasSameCompatibilityType(TypeSerializerSchemaCompatibility<T> expectedCompatibilty) {
return new TypeSafeMatcher<TypeSerializerSchemaCompatibility<T>>() {
@Override
@@ -451,9 +451,9 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
if (expectedCompatibilty.isCompatibleAsIs()) {
return testResultCompatibility.isCompatibleAsIs();
} else if (expectedCompatibilty.isIncompatible()) {
- return testResultCompatibility.isCompatibleAfterMigration();
- } else if (expectedCompatibilty.isIncompatible()) {
return testResultCompatibility.isIncompatible();
+ } else if (expectedCompatibilty.isCompatibleAfterMigration()) {
+ return testResultCompatibility.isCompatibleAfterMigration();
} else if (expectedCompatibilty.isCompatibleWithReconfiguredSerializer()) {
return testResultCompatibility.isCompatibleWithReconfiguredSerializer();
}
[flink] 02/11: [FLINK-11334] Remove old deserialization logic from
new EnumSerializer Snapshot
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 22f7aeff8717306403d98a17eddc6be6e21dd14c
Author: Igal Shilman <ig...@data-artisans.com>
AuthorDate: Tue Feb 19 16:33:10 2019 +0100
[FLINK-11334] Remove old deserialization logic from new EnumSerializer Snapshot
This code will never be used to deserialize from an old
SerializerConfigSnapshot.
---
.../api/common/typeutils/base/EnumSerializer.java | 24 ++++------------------
1 file changed, 4 insertions(+), 20 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index 73aea19..6312780 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -220,26 +220,10 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
enumClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
- if (readVersion == 1) {
- try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
- try {
- T[] legacyEnumConstants = InstantiationUtil.deserializeObject(inViewWrapper, userCodeClassLoader);
- this.previousEnumConstants = buildEnumConstantsList(legacyEnumConstants);
- } catch (ClassNotFoundException e) {
- throw new IOException("The requested enum class cannot be found in classpath.", e);
- } catch (IllegalArgumentException e) {
- throw new IOException("A previously existing enum constant of "
- + enumClass.getName() + " no longer exists.", e);
- }
- }
- } else if (readVersion >= 2) {
- int numEnumConstants = in.readInt();
- this.previousEnumConstants = new ArrayList<>(numEnumConstants);
- for (int i = 0; i < numEnumConstants; i++) {
- previousEnumConstants.add(in.readUTF());
- }
- } else {
- throw new IOException("Cannot deserialize EnumSerializerSnapshot with version " + readVersion);
+ int numEnumConstants = in.readInt();
+ this.previousEnumConstants = new ArrayList<>(numEnumConstants);
+ for (int i = 0; i < numEnumConstants; i++) {
+ previousEnumConstants.add(in.readUTF());
}
}
[flink] 09/11: [FLINK-11334] Fix
EnumSerializerSnapshotMigrationTest to check reconfiguration
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 754cdaec788743c4caa3dad22996fbde05e704e3
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Wed Feb 20 15:21:54 2019 +0100
[FLINK-11334] Fix EnumSerializerSnapshotMigrationTest to check reconfiguration
---
.../base/EnumSerializerSnapshotMigrationTest.java | 61 ++++++++++++++++++++--
1 file changed, 56 insertions(+), 5 deletions(-)
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
index f9e6955..ab4204e 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMigrationTest.java
@@ -18,14 +18,26 @@
package org.apache.flink.api.common.typeutils.base;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
import org.apache.flink.testutils.migration.MigrationVersion;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import java.util.Arrays;
import java.util.Collection;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.BAR;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.EMMA;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.FOO;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.NATHANIEL;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.PAULA;
+import static org.apache.flink.api.common.typeutils.base.TestEnum.PETER;
+
/**
* Migration tests for {@link EnumSerializer}.
*/
@@ -37,19 +49,58 @@ public class EnumSerializerSnapshotMigrationTest extends TypeSerializerSnapshotM
super(enumSerializer);
}
+ private static TestEnum[] previousEnumValues = {FOO, BAR, PETER, NATHANIEL, EMMA, PAULA};
+
@SuppressWarnings("unchecked")
@Parameterized.Parameters(name = "Test Specification = {0}")
public static Collection<TestSpecification<?>> testSpecifications() {
final TestSpecifications testSpecifications = new TestSpecifications(MigrationVersion.v1_6, MigrationVersion.v1_7);
- testSpecifications.add(
- SPEC_NAME,
- EnumSerializer.class,
- EnumSerializer.EnumSerializerSnapshot.class,
- () -> new EnumSerializer(TestEnum.class));
+ testSpecifications.addWithCompatibilityMatcher(
+ SPEC_NAME,
+ EnumSerializer.class,
+ EnumSerializer.EnumSerializerSnapshot.class,
+ () -> new EnumSerializer(TestEnum.class),
+ hasExpectedEnumSerializer(previousEnumValues));
return testSpecifications.get();
}
+
+
+ private static Matcher<TypeSerializerSchemaCompatibility<TestEnum>> hasExpectedEnumSerializer(
+ final TestEnum[] expectedEnumValues) {
+
+ return new TypeSafeDiagnosingMatcher<TypeSerializerSchemaCompatibility<TestEnum>>() {
+ @Override
+ protected boolean matchesSafely(
+ TypeSerializerSchemaCompatibility<TestEnum> item,
+ Description mismatchDescription) {
+ if (!item.isCompatibleWithReconfiguredSerializer()) {
+ mismatchDescription.appendText("compatibility mismatch ").appendValue(item);
+ return false;
+ }
+
+ EnumSerializer<TestEnum> reconfiguredSerialized =
+ (EnumSerializer<TestEnum>) item.getReconfiguredSerializer();
+
+ if (!Arrays.equals(reconfiguredSerialized.getValues(), expectedEnumValues)) {
+ mismatchDescription
+ .appendText("reconfigured values are ")
+ .appendValueList("{", ", ", "}", reconfiguredSerialized.getValues());
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description
+ .appendText("EnumSerializer with values ")
+ .appendValueList("{", ", ", "}", expectedEnumValues);
+ }
+ };
+ }
}