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 2018/10/30 13:57:21 UTC
[flink] 14/16: [FLINK-10716] Upgrade ListSerializer /
ArrayListSerializer for state evolution
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bcf14d5434ea16b1f17eb4acf67721bcad498a24
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Oct 29 20:26:52 2018 +0800
[FLINK-10716] Upgrade ListSerializer / ArrayListSerializer for state evolution
This upgrades their snapshot classes to implement the new interface, as
well as add their corresponding migration tests.
---
.../base/CollectionSerializerConfigSnapshot.java | 16 ++++
.../api/common/typeutils/base/ListSerializer.java | 31 +------
.../typeutils/base/ListSerializerSnapshot.java | 89 +++++++++++++++++++++
.../TypeSerializerSnapshotMigrationTestBase.java | 6 +-
.../base/ListSerializerSnapshotMigrationTest.java} | 38 ++++-----
.../test/resources/flink-1.6-list-serializer-data | Bin 0 -> 240 bytes
.../resources/flink-1.6-list-serializer-snapshot | Bin 0 -> 888 bytes
.../flink/runtime/state/ArrayListSerializer.java | 10 ++-
.../runtime/state/ArrayListSerializerSnapshot.java | 89 +++++++++++++++++++++
.../state/ArrayListSerializerMigrationTest.java | 40 ++++-----
.../state/StateBackendMigrationTestBase.java | 6 +-
.../resources/flink-1.6-arraylist-serializer-data | Bin 0 -> 240 bytes
.../flink-1.6-arraylist-serializer-snapshot | Bin 0 -> 881 bytes
13 files changed, 239 insertions(+), 86 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
index d1a3e95..377dd4c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.common.typeutils.base;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import java.util.Collection;
@@ -28,8 +29,11 @@ import java.util.Collection;
* Configuration snapshot of a serializer for collection types.
*
* @param <T> Type of the element.
+ *
+ * @deprecated this snapshot class should no longer be used by any serializers as their snapshot.
*/
@Internal
+@Deprecated
public final class CollectionSerializerConfigSnapshot<C extends Collection<T>, T>
extends CompositeTypeSerializerConfigSnapshot<C> {
@@ -43,6 +47,18 @@ public final class CollectionSerializerConfigSnapshot<C extends Collection<T>, T
}
@Override
+ public TypeSerializerSchemaCompatibility<C> resolveSchemaCompatibility(TypeSerializer<C> newSerializer) {
+ if (newSerializer instanceof ListSerializer) {
+ ListSerializerSnapshot<T> listSerializerSnapshot =
+ new ListSerializerSnapshot<>(((ListSerializer<T>) newSerializer).getElementSerializer());
+
+ return listSerializerSnapshot.resolveSchemaCompatibility((ListSerializer) newSerializer);
+ } else {
+ return super.resolveSchemaCompatibility(newSerializer);
+ }
+ }
+
+ @Override
public int getVersion() {
return VERSION;
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
index 7cda53f..ec2b542 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
@@ -19,14 +19,8 @@
package org.apache.flink.api.common.typeutils.base;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.CompatibilityResult;
-import org.apache.flink.api.common.typeutils.CompatibilityUtil;
-import org.apache.flink.api.common.typeutils.TypeDeserializerAdapter;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
-import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -181,29 +175,6 @@ public final class ListSerializer<T> extends TypeSerializer<List<T>> {
@Override
public TypeSerializerSnapshot<List<T>> snapshotConfiguration() {
- return new CollectionSerializerConfigSnapshot<>(elementSerializer);
- }
-
- @Override
- public CompatibilityResult<List<T>> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
- if (configSnapshot instanceof CollectionSerializerConfigSnapshot) {
- Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>> previousElemSerializerAndConfig =
- ((CollectionSerializerConfigSnapshot<?, ?>) configSnapshot).getSingleNestedSerializerAndConfig();
-
- CompatibilityResult<T> compatResult = CompatibilityUtil.resolveCompatibilityResult(
- previousElemSerializerAndConfig.f0,
- UnloadableDummyTypeSerializer.class,
- previousElemSerializerAndConfig.f1,
- elementSerializer);
-
- if (!compatResult.isRequiresMigration()) {
- return CompatibilityResult.compatible();
- } else if (compatResult.getConvertDeserializer() != null) {
- return CompatibilityResult.requiresMigration(
- new ListSerializer<>(new TypeDeserializerAdapter<>(compatResult.getConvertDeserializer())));
- }
- }
-
- return CompatibilityResult.requiresMigration();
+ return new ListSerializerSnapshot<>(elementSerializer);
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshot.java
new file mode 100644
index 0000000..2caf9fa
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshot.java
@@ -0,0 +1,89 @@
+/*
+ * 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.CompositeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Snapshot class for the {@link ListSerializer}.
+ */
+public class ListSerializerSnapshot<T> implements TypeSerializerSnapshot<List<T>> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ private CompositeSerializerSnapshot nestedElementSerializerSnapshot;
+
+ /**
+ * Constructor for read instantiation.
+ */
+ public ListSerializerSnapshot() {}
+
+ /**
+ * Constructor to create the snapshot for writing.
+ */
+ public ListSerializerSnapshot(TypeSerializer<T> elementSerializer) {
+ this.nestedElementSerializerSnapshot = new CompositeSerializerSnapshot(elementSerializer);
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public TypeSerializer<List<T>> restoreSerializer() {
+ return new ListSerializer<>(nestedElementSerializerSnapshot.getRestoreSerializer(0));
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<List<T>> resolveSchemaCompatibility(TypeSerializer<List<T>> newSerializer) {
+ checkState(nestedElementSerializerSnapshot != null);
+
+ if (newSerializer instanceof ListSerializer) {
+ ListSerializer<T> serializer = (ListSerializer<T>) newSerializer;
+
+ return nestedElementSerializerSnapshot.resolveCompatibilityWithNested(
+ TypeSerializerSchemaCompatibility.compatibleAsIs(),
+ serializer.getElementSerializer());
+ }
+ else {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ nestedElementSerializerSnapshot.writeCompositeSnapshot(out);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ this.nestedElementSerializerSnapshot = CompositeSerializerSnapshot.readCompositeSnapshot(in, userCodeClassLoader);
+ }
+}
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 64474e2..186f504 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
@@ -214,17 +214,17 @@ public abstract class TypeSerializerSnapshotMigrationTestBase<ElementT> extends
this.snapshotClass = snapshotClass;
}
- public TestSpecification withSerializerProvider(Supplier<? extends TypeSerializer<T>> serializerProvider) {
+ public TestSpecification<T> withSerializerProvider(Supplier<? extends TypeSerializer<T>> serializerProvider) {
this.serializerProvider = serializerProvider;
return this;
}
- public TestSpecification withSnapshotDataLocation(String snapshotDataLocation) {
+ public TestSpecification<T> withSnapshotDataLocation(String snapshotDataLocation) {
this.snapshotDataLocation = snapshotDataLocation;
return this;
}
- public TestSpecification withTestData(String testDataLocation, int testDataCount) {
+ public TestSpecification<T> withTestData(String testDataLocation, int testDataCount) {
this.testDataLocation = testDataLocation;
this.testDataCount = testDataCount;
return this;
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshotMigrationTest.java
similarity index 51%
copy from flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
copy to flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshotMigrationTest.java
index d1a3e95..02643fe 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/ListSerializerSnapshotMigrationTest.java
@@ -18,32 +18,24 @@
package org.apache.flink.api.common.typeutils.base;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
-import java.util.Collection;
+import java.util.List;
/**
- * Configuration snapshot of a serializer for collection types.
- *
- * @param <T> Type of the element.
+ * Migration test for the {@link ListSerializerSnapshot}.
*/
-@Internal
-public final class CollectionSerializerConfigSnapshot<C extends Collection<T>, T>
- extends CompositeTypeSerializerConfigSnapshot<C> {
-
- private static final int VERSION = 1;
-
- /** This empty nullary constructor is required for deserializing the configuration. */
- public CollectionSerializerConfigSnapshot() {}
-
- public CollectionSerializerConfigSnapshot(TypeSerializer<T> elementSerializer) {
- super(elementSerializer);
- }
-
- @Override
- public int getVersion() {
- return VERSION;
+public class ListSerializerSnapshotMigrationTest extends TypeSerializerSnapshotMigrationTestBase<List<String>> {
+
+ private static final String DATA = "flink-1.6-list-serializer-data";
+ private static final String SNAPSHOT = "flink-1.6-list-serializer-snapshot";
+
+ public ListSerializerSnapshotMigrationTest() {
+ super(
+ TestSpecification.<List<String>>builder("1.6-list-serializer", ListSerializer.class, ListSerializerSnapshot.class)
+ .withSerializerProvider(() -> new ListSerializer<>(StringSerializer.INSTANCE))
+ .withSnapshotDataLocation(SNAPSHOT)
+ .withTestData(DATA, 10)
+ );
}
}
diff --git a/flink-core/src/test/resources/flink-1.6-list-serializer-data b/flink-core/src/test/resources/flink-1.6-list-serializer-data
new file mode 100644
index 0000000..7b6c68a
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-list-serializer-data differ
diff --git a/flink-core/src/test/resources/flink-1.6-list-serializer-snapshot b/flink-core/src/test/resources/flink-1.6-list-serializer-snapshot
new file mode 100644
index 0000000..8bb94b4
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-list-serializer-snapshot differ
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
index 3614450..6fa9f02 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializer.java
@@ -142,15 +142,19 @@ final public class ArrayListSerializer<T> extends TypeSerializer<ArrayList<T>> {
}
// --------------------------------------------------------------------------------------------
- // Serializer configuration snapshotting & compatibility
+ // Serializer snapshots
// --------------------------------------------------------------------------------------------
@Override
- public TypeSerializerConfigSnapshot<ArrayList<T>> snapshotConfiguration() {
- return new CollectionSerializerConfigSnapshot<>(elementSerializer);
+ public TypeSerializerSnapshot<ArrayList<T>> snapshotConfiguration() {
+ return new ArrayListSerializerSnapshot<>(elementSerializer);
}
+ /**
+ * NOTE: this method cannot be removed until {@link CollectionSerializerConfigSnapshot} is fully removed.
+ */
@Override
+ @SuppressWarnings("deprecation")
public CompatibilityResult<ArrayList<T>> ensureCompatibility(TypeSerializerConfigSnapshot<?> configSnapshot) {
if (configSnapshot instanceof CollectionSerializerConfigSnapshot) {
Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>> previousElemSerializerAndConfig =
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializerSnapshot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializerSnapshot.java
new file mode 100644
index 0000000..7fc8c51
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ArrayListSerializerSnapshot.java
@@ -0,0 +1,89 @@
+/*
+ * 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.runtime.state;
+
+import org.apache.flink.api.common.typeutils.CompositeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Snapshot class for the {@link ArrayListSerializer}.
+ */
+public class ArrayListSerializerSnapshot<T> implements TypeSerializerSnapshot<ArrayList<T>> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ private CompositeSerializerSnapshot nestedElementSerializerSnapshot;
+
+ /**
+ * Constructor for read instantiation.
+ */
+ public ArrayListSerializerSnapshot() {}
+
+ /**
+ * Constructor for creating the snapshot for writing.
+ */
+ public ArrayListSerializerSnapshot(TypeSerializer<T> elementSerializer) {
+ this.nestedElementSerializerSnapshot = new CompositeSerializerSnapshot(elementSerializer);
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public TypeSerializer<ArrayList<T>> restoreSerializer() {
+ return new ArrayListSerializer<>(nestedElementSerializerSnapshot.getRestoreSerializer(0));
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<ArrayList<T>> resolveSchemaCompatibility(TypeSerializer<ArrayList<T>> newSerializer) {
+ checkState(nestedElementSerializerSnapshot != null);
+
+ if (newSerializer instanceof ArrayListSerializer) {
+ ArrayListSerializer<T> serializer = (ArrayListSerializer<T>) newSerializer;
+
+ return nestedElementSerializerSnapshot.resolveCompatibilityWithNested(
+ TypeSerializerSchemaCompatibility.compatibleAsIs(),
+ serializer.getElementSerializer());
+ }
+ else {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ nestedElementSerializerSnapshot.writeCompositeSnapshot(out);
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ this.nestedElementSerializerSnapshot = CompositeSerializerSnapshot.readCompositeSnapshot(in, userCodeClassLoader);
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ArrayListSerializerMigrationTest.java
similarity index 51%
copy from flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
copy to flink-runtime/src/test/java/org/apache/flink/runtime/state/ArrayListSerializerMigrationTest.java
index d1a3e95..b355795 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/CollectionSerializerConfigSnapshot.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ArrayListSerializerMigrationTest.java
@@ -16,34 +16,28 @@
* limitations under the License.
*/
-package org.apache.flink.api.common.typeutils.base;
+package org.apache.flink.runtime.state;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import java.util.Collection;
+import java.util.ArrayList;
/**
- * Configuration snapshot of a serializer for collection types.
- *
- * @param <T> Type of the element.
+ * Migration test for the {@link ArrayListSerializerSnapshot}.
*/
-@Internal
-public final class CollectionSerializerConfigSnapshot<C extends Collection<T>, T>
- extends CompositeTypeSerializerConfigSnapshot<C> {
-
- private static final int VERSION = 1;
-
- /** This empty nullary constructor is required for deserializing the configuration. */
- public CollectionSerializerConfigSnapshot() {}
-
- public CollectionSerializerConfigSnapshot(TypeSerializer<T> elementSerializer) {
- super(elementSerializer);
+public class ArrayListSerializerMigrationTest extends TypeSerializerSnapshotMigrationTestBase<ArrayList<String>> {
+
+ private static final String DATA = "flink-1.6-arraylist-serializer-data";
+ private static final String SNAPSHOT = "flink-1.6-arraylist-serializer-snapshot";
+
+ public ArrayListSerializerMigrationTest() {
+ super(
+ TestSpecification.<ArrayList<String>>builder("1.6-arraylist-serializer", ArrayListSerializer.class, ArrayListSerializerSnapshot.class)
+ .withSerializerProvider(() -> new ArrayListSerializer<>(StringSerializer.INSTANCE))
+ .withSnapshotDataLocation(SNAPSHOT)
+ .withTestData(DATA, 10)
+ );
}
- @Override
- public int getVersion() {
- return VERSION;
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
index de510de..f5f30d5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java
@@ -39,7 +39,6 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.StateMigrationException;
import org.apache.flink.util.TestLogger;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -141,7 +140,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
}
}
- @Ignore("This won't pass until the ListSerializer snapshot has been upgraded to the new TypeSerializerSnapshot interface")
@Test
public void testKeyedListStateMigration() throws Exception {
CheckpointStreamFactory streamFactory = createStreamFactory();
@@ -198,14 +196,14 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend.setCurrentKey(2);
Iterator<TestType> iterable2 = listState.get().iterator();
Assert.assertEquals(new TestType("key-2", 1), iterable2.next());
- Assert.assertFalse(iterable1.hasNext());
+ Assert.assertFalse(iterable2.hasNext());
listState.add(new TestType("new-key-2", 456));
backend.setCurrentKey(3);
Iterator<TestType> iterable3 = listState.get().iterator();
Assert.assertEquals(new TestType("key-3", 1), iterable3.next());
Assert.assertEquals(new TestType("key-3", 2), iterable3.next());
- Assert.assertFalse(iterable1.hasNext());
+ Assert.assertFalse(iterable3.hasNext());
listState.add(new TestType("new-key-3", 777));
} finally {
backend.dispose();
diff --git a/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-data b/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-data
new file mode 100644
index 0000000..7b6c68a
Binary files /dev/null and b/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-data differ
diff --git a/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-snapshot b/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-snapshot
new file mode 100644
index 0000000..f6486ae
Binary files /dev/null and b/flink-runtime/src/test/resources/flink-1.6-arraylist-serializer-snapshot differ