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 2017/06/20 12:38:21 UTC
[1/3] flink git commit: [FLINK-6922] [serializer] Remove Java
serialization from Enum(Value)SerializerConfigSnapshot
Repository: flink
Updated Branches:
refs/heads/master b9cce5167 -> f24a499b3
[FLINK-6922] [serializer] Remove Java serialization from Enum(Value)SerializerConfigSnapshot
This commit removes the use of Java serialization for serializing the
enum class and constants in the Java EnumSerializerConfigSnapshot and
Scala ScalaEnumSerializerConfigSnapshot.
This closes #4147.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f24a499b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f24a499b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f24a499b
Branch: refs/heads/master
Commit: f24a499b37515b43789ae23d83c1cd9eb2f827b3
Parents: 228faf8
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Tue Jun 20 18:17:21 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Jun 20 20:37:13 2017 +0800
----------------------------------------------------------------------
.../common/typeutils/base/EnumSerializer.java | 123 +++++++++++-----
.../base/EnumSerializerUpgradeTest.java | 144 +++++++++++++++++++
.../scala/typeutils/EnumValueSerializer.scala | 81 +++++------
.../typeutils/EnumValueSerializerTest.scala | 2 -
.../EnumValueSerializerUpgradeTest.scala | 5 -
flink-tests/pom.xml | 8 --
.../_metadata | Bin 217183 -> 213759 bytes
.../_metadata | Bin 217183 -> 213759 bytes
.../_metadata | Bin 217183 -> 213759 bytes
.../_metadata | Bin 217183 -> 213759 bytes
10 files changed, 271 insertions(+), 92 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
----------------------------------------------------------------------
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 d9246ae..c40fefc 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
@@ -21,9 +21,9 @@ package org.apache.flink.api.common.typeutils.base;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.Array;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.HashMap;
-import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Map;
import org.apache.flink.annotation.Internal;
@@ -33,7 +33,6 @@ 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.java.typeutils.runtime.DataInputViewStream;
-import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.util.InstantiationUtil;
@@ -185,22 +184,47 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
if (enumClass.equals(config.getTypeClass())) {
- // reorder enum constants so that previously existing constants
- // remain in the same order, and new
- LinkedHashSet<T> reorderedEnumConstants = new LinkedHashSet<>();
- reorderedEnumConstants.addAll(Arrays.asList(config.getEnumConstants()));
- reorderedEnumConstants.addAll(Arrays.asList(enumClass.getEnumConstants()));
-
- // regenerate enum constant to ordinal bidirectional map
- this.values = (T[]) Array.newInstance(enumClass, reorderedEnumConstants.size());
- this.valueToOrdinal.clear();
- int i = 0;
- for (T constant : reorderedEnumConstants) {
- this.values[i] = constant;
- this.valueToOrdinal.put(constant, i);
- i++;
+ T[] reorderedEnumConstants = (T[]) Array.newInstance(enumClass, this.values.length);
+ Map<T, Integer> rebuiltEnumConstantToOrdinalMap = new HashMap<>(this.values.length);
+
+ List<String> previousEnumConstants = config.getEnumConstants();
+
+ if (previousEnumConstants.size() <= this.values.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 CompatibilityResult.requiresMigration();
+ }
+ }
+ } else {
+ // some enum constants have been removed (because there are
+ // fewer constants now), and therefore requires migration
+ return CompatibilityResult.requiresMigration();
}
+ // 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++;
+ }
+ }
+ }
+
+ // if we reach here, we can simply reconfigure ourselves to be compatible
+ this.values = reorderedEnumConstants;
+ this.valueToOrdinal = rebuiltEnumConstantToOrdinalMap;
return CompatibilityResult.compatible();
}
}
@@ -219,24 +243,25 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
public static final class EnumSerializerConfigSnapshot<T extends Enum<T>>
extends GenericTypeSerializerConfigSnapshot<T> {
- private static final int VERSION = 1;
+ private static final int VERSION = 2;
- private T[] enumConstants;
+ private List<String> enumConstants;
/** This empty nullary constructor is required for deserializing the configuration. */
public EnumSerializerConfigSnapshot() {}
- public EnumSerializerConfigSnapshot(Class<T> enumClass, T[] enumConstants) {
+ public EnumSerializerConfigSnapshot(Class<T> enumClass, T[] enumConstantsArr) {
super(enumClass);
- this.enumConstants = Preconditions.checkNotNull(enumConstants);
+ this.enumConstants = buildEnumConstantsList(Preconditions.checkNotNull(enumConstantsArr));
}
@Override
public void write(DataOutputView out) throws IOException {
super.write(out);
- try (final DataOutputViewStream outViewWrapper = new DataOutputViewStream(out)) {
- InstantiationUtil.serializeObject(outViewWrapper, enumConstants);
+ out.writeInt(enumConstants.size());
+ for (String enumConstant : enumConstants) {
+ out.writeUTF(enumConstant);
}
}
@@ -244,15 +269,27 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
public void read(DataInputView in) throws IOException {
super.read(in);
- try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
- try {
- enumConstants = InstantiationUtil.deserializeObject(inViewWrapper, getUserCodeClassLoader());
- } 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 "
- + getTypeClass().getName() + " no longer exists.", e);
+ if (getReadVersion() == 1) {
+ try (final DataInputViewStream inViewWrapper = new DataInputViewStream(in)) {
+ try {
+ T[] legacyEnumConstants = InstantiationUtil.deserializeObject(inViewWrapper, getUserCodeClassLoader());
+ this.enumConstants = 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 "
+ + getTypeClass().getName() + " no longer exists.", e);
+ }
+ }
+ } else if (getReadVersion() == VERSION) {
+ int numEnumConstants = in.readInt();
+
+ this.enumConstants = new ArrayList<>(numEnumConstants);
+ for (int i = 0; i < numEnumConstants; i++) {
+ enumConstants.add(in.readUTF());
}
+ } else {
+ throw new IOException("Cannot deserialize EnumSerializerConfigSnapshot with version " + getReadVersion());
}
}
@@ -261,21 +298,33 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
return VERSION;
}
- public T[] getEnumConstants() {
+ @Override
+ public int[] getCompatibleVersions() {
+ return new int[] {VERSION, 1};
+ }
+
+ public List<String> getEnumConstants() {
return enumConstants;
}
@Override
public boolean equals(Object obj) {
- return super.equals(obj)
- && Arrays.equals(
- enumConstants,
- ((EnumSerializerConfigSnapshot) obj).getEnumConstants());
+ return super.equals(obj) && enumConstants.equals(((EnumSerializerConfigSnapshot) obj).getEnumConstants());
}
@Override
public int hashCode() {
- return super.hashCode() * 31 + Arrays.hashCode(enumConstants);
+ return super.hashCode() * 31 + enumConstants.hashCode();
+ }
+
+ 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;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..6502eb3
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerUpgradeTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.CompatibilityResult;
+import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSerializationUtil;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.util.TestLogger;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import javax.tools.JavaCompiler;
+import javax.tools.ToolProvider;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+
+public class EnumSerializerUpgradeTest extends TestLogger {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static final String ENUM_NAME = "EnumSerializerUpgradeTestEnum";
+
+ private static final String ENUM_A = "public enum " + ENUM_NAME + " { A, B, C }";
+ private static final String ENUM_B = "public enum " + ENUM_NAME + " { A, B, C, D }";
+ private static final String ENUM_C = "public enum " + ENUM_NAME + " { A, C }";
+ private static final String ENUM_D = "public enum " + ENUM_NAME + " { A, C, B }";
+
+ /**
+ * Check that identical enums don't require migration
+ */
+ @Test
+ public void checkIndenticalEnums() throws Exception {
+ Assert.assertFalse(checkCompatibility(ENUM_A, ENUM_A).isRequiresMigration());
+ }
+
+ /**
+ * Check that appending fields to the enum does not require migration
+ */
+ @Test
+ public void checkAppendedField() throws Exception {
+ Assert.assertFalse(checkCompatibility(ENUM_A, ENUM_B).isRequiresMigration());
+ }
+
+ /**
+ * Check that removing enum fields requires migration
+ */
+ @Test
+ public void checkRemovedField() throws Exception {
+ Assert.assertTrue(checkCompatibility(ENUM_A, ENUM_C).isRequiresMigration());
+ }
+
+ /**
+ * Check that changing the enum field order don't require migration
+ */
+ @Test
+ public void checkDifferentFieldOrder() throws Exception {
+ Assert.assertFalse(checkCompatibility(ENUM_A, ENUM_D).isRequiresMigration());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static CompatibilityResult checkCompatibility(String enumSourceA, String enumSourceB)
+ throws IOException, ClassNotFoundException {
+
+ ClassLoader classLoader = compileAndLoadEnum(
+ temporaryFolder.newFolder(), ENUM_NAME + ".java", enumSourceA);
+
+ EnumSerializer enumSerializer = new EnumSerializer(classLoader.loadClass(ENUM_NAME));
+
+ TypeSerializerConfigSnapshot snapshot = enumSerializer.snapshotConfiguration();
+ byte[] snapshotBytes;
+ try (
+ ByteArrayOutputStream outBuffer = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper outputViewStreamWrapper = new DataOutputViewStreamWrapper(outBuffer)) {
+
+ TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(outputViewStreamWrapper, snapshot);
+ snapshotBytes = outBuffer.toByteArray();
+ }
+
+ ClassLoader classLoader2 = compileAndLoadEnum(
+ temporaryFolder.newFolder(), ENUM_NAME + ".java", enumSourceB);
+
+ TypeSerializerConfigSnapshot restoredSnapshot;
+ try (
+ ByteArrayInputStream inBuffer = new ByteArrayInputStream(snapshotBytes);
+ DataInputViewStreamWrapper inputViewStreamWrapper = new DataInputViewStreamWrapper(inBuffer)) {
+
+ restoredSnapshot = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(inputViewStreamWrapper, classLoader2);
+ }
+
+ EnumSerializer enumSerializer2 = new EnumSerializer(classLoader2.loadClass(ENUM_NAME));
+ return enumSerializer2.ensureCompatibility(restoredSnapshot);
+ }
+
+ private static ClassLoader compileAndLoadEnum(File root, String filename, String source) throws IOException {
+ File file = writeSourceFile(root, filename, source);
+
+ compileClass(file);
+
+ return new URLClassLoader(
+ new URL[]{root.toURI().toURL()},
+ Thread.currentThread().getContextClassLoader());
+ }
+
+ private static File writeSourceFile(File root, String filename, String source) throws IOException {
+ File file = new File(root, filename);
+ FileWriter fileWriter = new FileWriter(file);
+
+ fileWriter.write(source);
+ fileWriter.close();
+
+ return file;
+ }
+
+ private static int compileClass(File sourceFile) {
+ JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
+ return compiler.run(null, null, null, sourceFile.getPath());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
index 344b56d..b7ca32d 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
@@ -90,24 +90,22 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
case enumSerializerConfigSnapshot: EnumValueSerializer.ScalaEnumSerializerConfigSnapshot[_] =>
val enumClass = enum.getClass.asInstanceOf[Class[E]]
if (enumClass.equals(enumSerializerConfigSnapshot.getEnumClass)) {
- val previousEnumConstants:List[(String, Int)] =
+ val previousEnumConstants: List[(String, Int)] =
enumSerializerConfigSnapshot.getEnumConstants
- if (previousEnumConstants != null) {
- for ((previousEnumConstant, idx) <- previousEnumConstants) {
- val enumValue = try {
- enum(idx)
- } catch {
- case _: NoSuchElementException =>
- // couldn't find an enum value for the given index
- return CompatibilityResult.requiresMigration()
- }
-
- if (!previousEnumConstant.equals(enumValue.toString)) {
- // compatible only if new enum constants are only appended,
- // and original constants must be in the exact same order
+ for ((previousEnumConstant, idx) <- previousEnumConstants) {
+ val enumValue = try {
+ enum(idx)
+ } catch {
+ case _: NoSuchElementException =>
+ // couldn't find an enum value for the given index
return CompatibilityResult.requiresMigration()
- }
+ }
+
+ if (!previousEnumConstant.equals(enumValue.toString)) {
+ // compatible only if new enum constants are only appended,
+ // and original constants must be in the exact same order
+ return CompatibilityResult.requiresMigration()
}
}
@@ -141,7 +139,7 @@ object EnumValueSerializer {
try {
val outViewWrapper = new DataOutputViewStream(out)
try {
- InstantiationUtil.serializeObject(outViewWrapper, enumClass)
+ out.writeUTF(enumClass.getName)
out.writeInt(enumConstants.length)
for ((name, idx) <- enumConstants) {
@@ -157,34 +155,37 @@ object EnumValueSerializer {
try {
val inViewWrapper = new DataInputViewStream(in)
- try
- try {
+ try {
+ if (getReadVersion == 1) {
enumClass = InstantiationUtil.deserializeObject(
inViewWrapper, getUserCodeClassLoader)
- if (getReadVersion == 1) {
- // read null from input stream
- InstantiationUtil.deserializeObject(inViewWrapper, getUserCodeClassLoader)
- enumConstants = List()
- } else if (getReadVersion == 2) {
- val length = in.readInt()
- val listBuffer = ListBuffer[(String, Int)]()
-
- for (_ <- 0 until length) {
- val name = in.readUTF()
- val idx = in.readInt()
- listBuffer += ((name, idx))
- }
-
- enumConstants = listBuffer.toList
- } else {
- throw new IOException(s"Cannot deserialize ${getClass.getSimpleName} with version $getReadVersion.")
+ // read null from input stream
+ InstantiationUtil.deserializeObject(inViewWrapper, getUserCodeClassLoader)
+ enumConstants = List()
+ } else if (getReadVersion == ScalaEnumSerializerConfigSnapshot.VERSION) {
+ enumClass = Class.forName(
+ in.readUTF(), true, getUserCodeClassLoader).asInstanceOf[Class[E]]
+
+ val length = in.readInt()
+ val listBuffer = ListBuffer[(String, Int)]()
+
+ for (_ <- 0 until length) {
+ val name = in.readUTF()
+ val idx = in.readInt()
+ listBuffer += ((name, idx))
}
- } catch {
- case e: ClassNotFoundException =>
- throw new IOException("The requested enum class cannot be found in classpath.", e)
+
+ enumConstants = listBuffer.toList
+ } else {
+ throw new IOException(
+ s"Cannot deserialize ${getClass.getSimpleName} with version $getReadVersion.")
}
- finally if (inViewWrapper != null) inViewWrapper.close()
+ } catch {
+ case e: ClassNotFoundException =>
+ throw new IOException("The requested enum class cannot be found in classpath.", e)
+ }
+ finally if (inViewWrapper != null) inViewWrapper.close()
}
}
@@ -214,7 +215,7 @@ object EnumValueSerializer {
}
override def getCompatibleVersions: Array[Int] = {
- Array(1, 2)
+ Array(ScalaEnumSerializerConfigSnapshot.VERSION, 1)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
index 5009271..f89e9bc 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
@@ -28,8 +28,6 @@ class EnumValueSerializerTest extends TestLogger with JUnitSuiteLike {
/**
* Tests that the snapshot configuration can be created and that the serializer
* is compatible when being called with the created serializer snapshot
- *
- * FLINK-6914
*/
@Test
def testEnumValueSerializerEnsureCompatibilityIdempotency() {
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
index af725f6..8be4deb 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
@@ -44,7 +44,6 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
val enumA =
s"""
- |@SerialVersionUID(1L)
|object $enumName extends Enumeration {
| val A, B, C = Value
|}
@@ -52,7 +51,6 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
val enumB =
s"""
- |@SerialVersionUID(1L)
|object $enumName extends Enumeration {
| val A, B, C, D = Value
|}
@@ -60,7 +58,6 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
val enumC =
s"""
- |@SerialVersionUID(1L)
|object $enumName extends Enumeration {
| val A, C = Value
|}
@@ -68,7 +65,6 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
val enumD =
s"""
- |@SerialVersionUID(1L)
|object $enumName extends Enumeration {
| val A, C, B = Value
|}
@@ -76,7 +72,6 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
val enumE =
s"""
- |@SerialVersionUID(1L)
|object $enumName extends Enumeration {
| val A = Value(42)
| val B = Value(5)
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml
index 4caf8a6..beac803 100644
--- a/flink-tests/pom.xml
+++ b/flink-tests/pom.xml
@@ -120,14 +120,6 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
index c18fd09..4e68685 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
index 301725b..6c9e433 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
index e154243..e0c6da6 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/f24a499b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
index 08aa333..f719eac 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata differ
[2/3] flink git commit: [FLINK-6948] [serializer] Harden
EnumValueSerializer to detect changed enum indices
Posted by tz...@apache.org.
[FLINK-6948] [serializer] Harden EnumValueSerializer to detect changed enum indices
This PR changes the seriailization format of the ScalaEnumSerializerConfigSnapshot to also include the
ordinal value of an enum value when being deserialized. This allows to detect if the ordinal values
have been changed and, thus, if migration is required.
IMPORTANT: This PR changes the serialization format of ScalaEnumSerializerConfigSnapshot.
Remove backwards compatibility path for 1.3.1
This closes #4142.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/228faf8c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/228faf8c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/228faf8c
Branch: refs/heads/master
Commit: 228faf8c2a113de0dea366668e87484dbe7ec8b5
Parents: e520023
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Jun 19 12:49:05 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Jun 20 20:37:13 2017 +0800
----------------------------------------------------------------------
.../scala/typeutils/EnumValueSerializer.scala | 65 ++++++++++++++-----
.../src/test/resources/log4j-test.properties | 31 +++++++++
.../EnumValueSerializerUpgradeTest.scala | 20 ++++++
.../_metadata | Bin 218895 -> 217183 bytes
.../_metadata | Bin 218895 -> 217183 bytes
.../_metadata | Bin 218119 -> 217183 bytes
.../_metadata | Bin 218119 -> 217183 bytes
7 files changed, 99 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
index 8be5e0b..344b56d 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
@@ -26,6 +26,8 @@ import org.apache.flink.api.java.typeutils.runtime.{DataInputViewStream, DataOut
import org.apache.flink.core.memory.{DataInputView, DataOutputView}
import org.apache.flink.util.{InstantiationUtil, Preconditions}
+import scala.collection.mutable.ListBuffer
+
/**
* Serializer for [[Enumeration]] values.
*/
@@ -88,18 +90,23 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
case enumSerializerConfigSnapshot: EnumValueSerializer.ScalaEnumSerializerConfigSnapshot[_] =>
val enumClass = enum.getClass.asInstanceOf[Class[E]]
if (enumClass.equals(enumSerializerConfigSnapshot.getEnumClass)) {
- val previousEnumConstants = enumSerializerConfigSnapshot.getEnumConstants
+ val previousEnumConstants:List[(String, Int)] =
+ enumSerializerConfigSnapshot.getEnumConstants
if (previousEnumConstants != null) {
- for (i <- enum.values.iterator) {
- // skip the check for all newly added fields
- if (i.id < previousEnumConstants.length) {
- if (!previousEnumConstants(i.id).equals(i.toString)) {
- // compatible only if new enum constants are only appended,
- // and original constants must be in the exact same order
-
+ for ((previousEnumConstant, idx) <- previousEnumConstants) {
+ val enumValue = try {
+ enum(idx)
+ } catch {
+ case _: NoSuchElementException =>
+ // couldn't find an enum value for the given index
return CompatibilityResult.requiresMigration()
- }
+ }
+
+ if (!previousEnumConstant.equals(enumValue.toString)) {
+ // compatible only if new enum constants are only appended,
+ // and original constants must be in the exact same order
+ return CompatibilityResult.requiresMigration()
}
}
}
@@ -120,12 +127,12 @@ object EnumValueSerializer {
extends TypeSerializerConfigSnapshot {
var enumClass: Class[E] = _
- var enumConstants: List[String] = _
+ var enumConstants: List[(String, Int)] = _
def this(enum: E) = {
this()
this.enumClass = Preconditions.checkNotNull(enum).getClass.asInstanceOf[Class[E]]
- this.enumConstants = enum.values.toList.map(_.toString)
+ this.enumConstants = enum.values.toList.map(x => (x.toString, x.id))
}
override def write(out: DataOutputView): Unit = {
@@ -135,7 +142,12 @@ object EnumValueSerializer {
val outViewWrapper = new DataOutputViewStream(out)
try {
InstantiationUtil.serializeObject(outViewWrapper, enumClass)
- InstantiationUtil.serializeObject(outViewWrapper, enumConstants)
+
+ out.writeInt(enumConstants.length)
+ for ((name, idx) <- enumConstants) {
+ out.writeUTF(name)
+ out.writeInt(idx)
+ }
} finally if (outViewWrapper != null) outViewWrapper.close()
}
}
@@ -150,8 +162,24 @@ object EnumValueSerializer {
enumClass = InstantiationUtil.deserializeObject(
inViewWrapper, getUserCodeClassLoader)
- enumConstants = InstantiationUtil.deserializeObject(
- inViewWrapper, getUserCodeClassLoader)
+ if (getReadVersion == 1) {
+ // read null from input stream
+ InstantiationUtil.deserializeObject(inViewWrapper, getUserCodeClassLoader)
+ enumConstants = List()
+ } else if (getReadVersion == 2) {
+ val length = in.readInt()
+ val listBuffer = ListBuffer[(String, Int)]()
+
+ for (_ <- 0 until length) {
+ val name = in.readUTF()
+ val idx = in.readInt()
+ listBuffer += ((name, idx))
+ }
+
+ enumConstants = listBuffer.toList
+ } else {
+ throw new IOException(s"Cannot deserialize ${getClass.getSimpleName} with version $getReadVersion.")
+ }
} catch {
case e: ClassNotFoundException =>
throw new IOException("The requested enum class cannot be found in classpath.", e)
@@ -164,7 +192,7 @@ object EnumValueSerializer {
def getEnumClass: Class[E] = enumClass
- def getEnumConstants: List[String] = enumConstants
+ def getEnumConstants: List[(String, Int)] = enumConstants
override def equals(obj: scala.Any): Boolean = {
if (obj == this) {
@@ -184,10 +212,13 @@ object EnumValueSerializer {
override def hashCode(): Int = {
enumClass.hashCode() * 31 + enumConstants.hashCode()
}
+
+ override def getCompatibleVersions: Array[Int] = {
+ Array(1, 2)
+ }
}
object ScalaEnumSerializerConfigSnapshot {
- val VERSION = 1
+ val VERSION = 2
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-scala/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/resources/log4j-test.properties b/flink-scala/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000..10792cd
--- /dev/null
+++ b/flink-scala/src/test/resources/log4j-test.properties
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+log4j.rootLogger=OFF, testlogger
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target = System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+
+# suppress the irrelevant (wrong) warnings from the netty channel handler
+log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR
+log4j.logger.org.apache.zookeeper=OFF
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
index c470cd0..af725f6 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
@@ -74,6 +74,16 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
|}
""".stripMargin
+ val enumE =
+ s"""
+ |@SerialVersionUID(1L)
+ |object $enumName extends Enumeration {
+ | val A = Value(42)
+ | val B = Value(5)
+ | val C = Value(1337)
+ |}
+ """.stripMargin
+
/**
* Check that identical enums don't require migration
*/
@@ -106,6 +116,16 @@ class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
assertTrue(checkCompatibility(enumA, enumD).isRequiresMigration)
}
+ /**
+ * Check that changing the enum ids causes a migration
+ */
+ @Test
+ def checkDifferentIds(): Unit = {
+ assertTrue(
+ "Different ids should cause a migration.",
+ checkCompatibility(enumA, enumE).isRequiresMigration)
+ }
+
def checkCompatibility(enumSourceA: String, enumSourceB: String)
: CompatibilityResult[Enumeration#Value] = {
import EnumValueSerializerUpgradeTest._
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
index f4db7ff..c18fd09 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
index 459b89e..301725b 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.10-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata
index eb87c0b..e154243 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-jobmanager-savepoint/_metadata differ
http://git-wip-us.apache.org/repos/asf/flink/blob/228faf8c/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata
index 6a33832..08aa333 100644
Binary files a/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/stateful-scala2.11-udf-migration-itcase-flink1.3-rocksdb-savepoint/_metadata differ
[3/3] flink git commit: [FLINK-6921] [serializer] Allow
EnumValueSerializer to deal with appended enum values
Posted by tz...@apache.org.
[FLINK-6921] [serializer] Allow EnumValueSerializer to deal with appended enum values
The problem was that we don't check the bounds of the array with the enum names contained
in the ScalaEnumSerializerConfigSnapshot.
This PR also adds an Enumeration upgrade test which makes sure that appended fields are
supported without migration. Moreover, it checks that a field removal and an order change
leads to a required migration.
This closes #4126.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e520023a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e520023a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e520023a
Branch: refs/heads/master
Commit: e520023ac768ac739537f7aa563f13aa208949ff
Parents: b9cce51
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Jun 14 16:53:49 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Jun 20 20:37:13 2017 +0800
----------------------------------------------------------------------
flink-scala/pom.xml | 10 +
.../scala/typeutils/EnumValueSerializer.scala | 13 +-
.../typeutils/EnumValueSerializerTest.scala | 46 +++++
.../EnumValueSerializerUpgradeTest.scala | 199 +++++++++++++++++++
4 files changed, 263 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e520023a/flink-scala/pom.xml
----------------------------------------------------------------------
diff --git a/flink-scala/pom.xml b/flink-scala/pom.xml
index 1c4d7c4..2df526e 100644
--- a/flink-scala/pom.xml
+++ b/flink-scala/pom.xml
@@ -107,6 +107,16 @@ under the License.
<artifactId>japicmp-maven-plugin</artifactId>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <!-- We have to disable this option so that the system class loader contains all
+ dependencies as URLs -->
+ <useManifestOnlyJar>false</useManifestOnlyJar>
+ </configuration>
+ </plugin>
+
<!-- Scala Compiler -->
<plugin>
<groupId>net.alchim31.maven</groupId>
http://git-wip-us.apache.org/repos/asf/flink/blob/e520023a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
index 119db93..8be5e0b 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
@@ -92,11 +92,14 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
if (previousEnumConstants != null) {
for (i <- enum.values.iterator) {
- if (!previousEnumConstants(i.id).equals(i.toString)) {
- // compatible only if new enum constants are only appended,
- // and original constants must be in the exact same order
-
- return CompatibilityResult.requiresMigration()
+ // skip the check for all newly added fields
+ if (i.id < previousEnumConstants.length) {
+ if (!previousEnumConstants(i.id).equals(i.toString)) {
+ // compatible only if new enum constants are only appended,
+ // and original constants must be in the exact same order
+
+ return CompatibilityResult.requiresMigration()
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e520023a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
new file mode 100644
index 0000000..5009271
--- /dev/null
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerTest.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.scala.typeutils
+
+import org.apache.flink.util.TestLogger
+import org.junit.Test
+import org.junit.Assert._
+import org.scalatest.junit.JUnitSuiteLike
+
+class EnumValueSerializerTest extends TestLogger with JUnitSuiteLike {
+
+ /**
+ * Tests that the snapshot configuration can be created and that the serializer
+ * is compatible when being called with the created serializer snapshot
+ *
+ * FLINK-6914
+ */
+ @Test
+ def testEnumValueSerializerEnsureCompatibilityIdempotency() {
+ val enumSerializer = new EnumValueSerializer(Letters)
+
+ val snapshot = enumSerializer.snapshotConfiguration()
+
+ assertFalse(enumSerializer.ensureCompatibility(snapshot).isRequiresMigration)
+ }
+}
+
+object Letters extends Enumeration {
+ val A, B, C = Value
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e520023a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
new file mode 100644
index 0000000..c470cd0
--- /dev/null
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializerUpgradeTest.scala
@@ -0,0 +1,199 @@
+/*
+ * 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.scala.typeutils
+
+import java.io._
+import java.net.{URL, URLClassLoader}
+
+import org.apache.flink.api.common.typeutils.{CompatibilityResult, TypeSerializerSerializationUtil}
+import org.apache.flink.core.memory.{DataInputViewStreamWrapper, DataOutputViewStreamWrapper}
+import org.apache.flink.util.TestLogger
+import org.junit.rules.TemporaryFolder
+import org.junit.{Rule, Test}
+import org.junit.Assert._
+import org.scalatest.junit.JUnitSuiteLike
+
+import scala.reflect.NameTransformer
+import scala.tools.nsc.reporters.ConsoleReporter
+import scala.tools.nsc.{GenericRunnerSettings, Global}
+
+class EnumValueSerializerUpgradeTest extends TestLogger with JUnitSuiteLike {
+
+ private val _tempFolder = new TemporaryFolder()
+
+ @Rule
+ def tempFolder = _tempFolder
+
+ val enumName = "EnumValueSerializerUpgradeTestEnum"
+
+ val enumA =
+ s"""
+ |@SerialVersionUID(1L)
+ |object $enumName extends Enumeration {
+ | val A, B, C = Value
+ |}
+ """.stripMargin
+
+ val enumB =
+ s"""
+ |@SerialVersionUID(1L)
+ |object $enumName extends Enumeration {
+ | val A, B, C, D = Value
+ |}
+ """.stripMargin
+
+ val enumC =
+ s"""
+ |@SerialVersionUID(1L)
+ |object $enumName extends Enumeration {
+ | val A, C = Value
+ |}
+ """.stripMargin
+
+ val enumD =
+ s"""
+ |@SerialVersionUID(1L)
+ |object $enumName extends Enumeration {
+ | val A, C, B = Value
+ |}
+ """.stripMargin
+
+ /**
+ * Check that identical enums don't require migration
+ */
+ @Test
+ def checkIdenticalEnums(): Unit = {
+ assertFalse(checkCompatibility(enumA, enumA).isRequiresMigration)
+ }
+
+ /**
+ * Check that appending fields to the enum does not require migration
+ */
+ @Test
+ def checkAppendedField(): Unit = {
+ assertFalse(checkCompatibility(enumA, enumB).isRequiresMigration)
+ }
+
+ /**
+ * Check that removing enum fields requires migration
+ */
+ @Test
+ def checkRemovedField(): Unit = {
+ assertTrue(checkCompatibility(enumA, enumC).isRequiresMigration)
+ }
+
+ /**
+ * Check that changing the enum field order requires migration
+ */
+ @Test
+ def checkDifferentFieldOrder(): Unit = {
+ assertTrue(checkCompatibility(enumA, enumD).isRequiresMigration)
+ }
+
+ def checkCompatibility(enumSourceA: String, enumSourceB: String)
+ : CompatibilityResult[Enumeration#Value] = {
+ import EnumValueSerializerUpgradeTest._
+
+ val classLoader = compileAndLoadEnum(tempFolder.newFolder(), s"$enumName.scala", enumSourceA)
+
+ val enum = instantiateEnum[Enumeration](classLoader, enumName)
+
+ val enumValueSerializer = new EnumValueSerializer(enum)
+ val snapshot = enumValueSerializer.snapshotConfiguration()
+
+ val baos = new ByteArrayOutputStream()
+ val output = new DataOutputViewStreamWrapper(baos)
+ TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(output, snapshot)
+
+ output.close()
+ baos.close()
+
+ val bais = new ByteArrayInputStream(baos.toByteArray)
+ val input= new DataInputViewStreamWrapper(bais)
+
+ val classLoader2 = compileAndLoadEnum(tempFolder.newFolder(), s"$enumName.scala", enumSourceB)
+
+ val snapshot2 = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
+ input,
+ classLoader2)
+ val enum2 = instantiateEnum[Enumeration](classLoader2, enumName)
+
+ val enumValueSerializer2 = new EnumValueSerializer(enum2)
+ enumValueSerializer2.ensureCompatibility(snapshot2)
+ }
+}
+
+object EnumValueSerializerUpgradeTest {
+ def compileAndLoadEnum(root: File, filename: String, source: String): ClassLoader = {
+ val file = writeSourceFile(root, filename, source)
+
+ compileScalaFile(file)
+
+ new URLClassLoader(
+ Array[URL](root.toURI.toURL),
+ Thread.currentThread().getContextClassLoader)
+ }
+
+ def instantiateEnum[T <: Enumeration](classLoader: ClassLoader, enumName: String): T = {
+ val clazz = classLoader.loadClass(enumName + "$").asInstanceOf[Class[_ <: Enumeration]]
+ val field = clazz.getField(NameTransformer.MODULE_INSTANCE_NAME)
+
+ field.get(null).asInstanceOf[T]
+ }
+
+ def writeSourceFile(root: File, filename: String, source: String): File = {
+ val file = new File(root, filename)
+ val fileWriter = new FileWriter(file)
+
+ fileWriter.write(source)
+
+ fileWriter.close()
+
+ file
+ }
+
+ def compileScalaFile(file: File): Unit = {
+ val in = new BufferedReader(new StringReader(""))
+ val out = new PrintWriter(new BufferedWriter(
+ new OutputStreamWriter(System.out)))
+
+ val settings = new GenericRunnerSettings(out.println _)
+
+ val classLoader = Thread.currentThread().getContextClassLoader
+
+ val urls = classLoader match {
+ case urlClassLoader: URLClassLoader =>
+ urlClassLoader.getURLs
+ case x => throw new IllegalStateException(s"Not possible to extract URLs " +
+ s"from class loader $x.")
+ }
+
+ settings.classpath.value = urls.map(_.toString).mkString(java.io.File.pathSeparator)
+ settings.outdir.value = file.getParent
+
+ val reporter = new ConsoleReporter(settings)
+ val global = new Global(settings, reporter)
+ val run = new global.Run
+
+ run.compile(List(file.getAbsolutePath))
+
+ reporter.printSummary()
+ }
+}
+