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/05/07 06:26:32 UTC
[2/8] flink git commit: [FLINK-9169] [runtime] Allow specifiying
serializer presence requirement in KeyedBackendSerializationProxy
[FLINK-9169] [runtime] Allow specifiying serializer presence requirement in KeyedBackendSerializationProxy
This commit consolidates logic of whether old serializers are required
to be present at restore time in the KeyedBackendSerializationProxy, via
a isSerializerPresenceRequired flag.
Heap-based backends typically set this to true, while RocksDB state
backend will set this to false. If set to true, restored serializers
cannot be the UnloadableDummyTypeSerializer, otherwise an IOException
will be thrown to fail the restore.
This closes #5950.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a7ed135f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a7ed135f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a7ed135f
Branch: refs/heads/master
Commit: a7ed135f5bd7c139ff756ee15583e918978116dd
Parents: 7c541a1
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Thu May 3 15:30:55 2018 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Mon May 7 10:56:22 2018 +0800
----------------------------------------------------------------------
.../state/KeyedBackendSerializationProxy.java | 35 +++++++--
.../state/heap/HeapKeyedStateBackend.java | 21 +-----
.../runtime/state/FileStateBackendTest.java | 5 ++
.../runtime/state/MemoryStateBackendTest.java | 5 ++
.../runtime/state/SerializationProxiesTest.java | 7 +-
.../runtime/state/StateBackendTestBase.java | 77 ++++++++++++++++++++
.../state/RocksDBKeyedStateBackend.java | 12 ++-
.../state/RocksDBStateBackendTest.java | 5 ++
8 files changed, 139 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
index 30b7344..f32e646 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
import org.apache.flink.api.common.typeutils.TypeSerializerSerializationUtil;
+import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.io.VersionedIOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
@@ -44,6 +45,9 @@ public class KeyedBackendSerializationProxy<K> extends VersionedIOReadableWritab
/** This specifies if we use a compressed format write the key-groups */
private boolean usingKeyGroupCompression;
+ /** This specifies whether or not to use dummy {@link UnloadableDummyTypeSerializer} when serializers cannot be read. */
+ private boolean isSerializerPresenceRequired;
+
private TypeSerializer<K> keySerializer;
private TypeSerializerConfigSnapshot keySerializerConfigSnapshot;
@@ -51,8 +55,9 @@ public class KeyedBackendSerializationProxy<K> extends VersionedIOReadableWritab
private ClassLoader userCodeClassLoader;
- public KeyedBackendSerializationProxy(ClassLoader userCodeClassLoader) {
+ public KeyedBackendSerializationProxy(ClassLoader userCodeClassLoader, boolean isSerializerPresenceRequired) {
this.userCodeClassLoader = Preconditions.checkNotNull(userCodeClassLoader);
+ this.isSerializerPresenceRequired = isSerializerPresenceRequired;
}
public KeyedBackendSerializationProxy(
@@ -139,17 +144,35 @@ public class KeyedBackendSerializationProxy<K> extends VersionedIOReadableWritab
this.keySerializer = (TypeSerializer<K>) keySerializerAndConfig.f0;
this.keySerializerConfigSnapshot = keySerializerAndConfig.f1;
} else {
- this.keySerializer = TypeSerializerSerializationUtil.tryReadSerializer(in, userCodeClassLoader);
+ this.keySerializer = TypeSerializerSerializationUtil.tryReadSerializer(in, userCodeClassLoader, true);
this.keySerializerConfigSnapshot = null;
}
+ if (isSerializerPresenceRequired) {
+ checkSerializerPresence(keySerializer);
+ }
+
int numKvStates = in.readShort();
stateMetaInfoSnapshots = new ArrayList<>(numKvStates);
for (int i = 0; i < numKvStates; i++) {
- stateMetaInfoSnapshots.add(
- KeyedBackendStateMetaInfoSnapshotReaderWriters
- .getReaderForVersion(getReadVersion(), userCodeClassLoader)
- .readStateMetaInfo(in));
+ RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?> snapshot = KeyedBackendStateMetaInfoSnapshotReaderWriters
+ .getReaderForVersion(getReadVersion(), userCodeClassLoader)
+ .readStateMetaInfo(in);
+
+ if (isSerializerPresenceRequired) {
+ checkSerializerPresence(snapshot.getNamespaceSerializer());
+ checkSerializerPresence(snapshot.getStateSerializer());
+ }
+ stateMetaInfoSnapshots.add(snapshot);
+ }
+ }
+
+ private void checkSerializerPresence(TypeSerializer<?> serializer) throws IOException {
+ if (serializer instanceof UnloadableDummyTypeSerializer) {
+ throw new IOException("Unable to restore keyed state, because a previous serializer" +
+ " of the keyed state is not present The serializer could have been removed from the classpath, " +
+ " or its implementation have changed and could not be loaded. This is a temporary restriction that will" +
+ " be fixed in future versions.");
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
index d0fb8a7..a02edb0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -74,7 +74,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
-import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
@@ -344,8 +343,10 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
try {
DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(fsDataInputStream);
+ // isSerializerPresenceRequired flag is set to true, since for the heap state backend,
+ // deserialization of state happens eagerly at restore time
KeyedBackendSerializationProxy<K> serializationProxy =
- new KeyedBackendSerializationProxy<>(userCodeClassLoader);
+ new KeyedBackendSerializationProxy<>(userCodeClassLoader, true);
serializationProxy.read(inView);
@@ -371,22 +372,6 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
serializationProxy.getStateMetaInfoSnapshots();
for (RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?> restoredMetaInfo : restoredMetaInfos) {
-
- if (restoredMetaInfo.getStateSerializer() == null ||
- restoredMetaInfo.getStateSerializer() instanceof UnloadableDummyTypeSerializer) {
-
- // must fail now if the previous serializer cannot be restored because there is no serializer
- // capable of reading previous state
- // TODO when eager state registration is in place, we can try to get a convert deserializer
- // TODO from the newly registered serializer instead of simply failing here
-
- throw new IOException("Unable to restore keyed state [" + restoredMetaInfo.getName() + "]." +
- " For memory-backed keyed state, the previous serializer of the keyed state must be" +
- " present; the serializer could have been removed from the classpath, or its implementation" +
- " have changed and could not be loaded. This is a temporary restriction that will be fixed" +
- " in future versions.");
- }
-
restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);
StateTable<K, ?, ?> stateTable = stateTables.get(restoredMetaInfo.getName());
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java
index ca74852..beea0c2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java
@@ -46,6 +46,11 @@ public class FileStateBackendTest extends StateBackendTestBase<FsStateBackend> {
return false;
}
+ @Override
+ protected boolean isSerializerPresenceRequiredOnRestore() {
+ return true;
+ }
+
// disable these because the verification does not work for this state backend
@Override
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java
index a13bc1d..0ba4c33 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java
@@ -59,6 +59,11 @@ public class MemoryStateBackendTest extends StateBackendTestBase<MemoryStateBack
return false;
}
+ @Override
+ protected boolean isSerializerPresenceRequiredOnRestore() {
+ return true;
+ }
+
// disable these because the verification does not work for this state backend
@Override
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
index 5176fed..3f78097 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
@@ -67,7 +67,7 @@ public class SerializationProxiesTest {
}
serializationProxy =
- new KeyedBackendSerializationProxy<>(Thread.currentThread().getContextClassLoader());
+ new KeyedBackendSerializationProxy<>(Thread.currentThread().getContextClassLoader(), true);
try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
serializationProxy.read(new DataInputViewStreamWrapper(in));
@@ -109,11 +109,14 @@ public class SerializationProxiesTest {
cnfThrowingSerializerClasses.add(LongSerializer.class.getName());
cnfThrowingSerializerClasses.add(DoubleSerializer.class.getName());
+ // we want to verify restore resilience when serializer presence is not required;
+ // set isSerializerPresenceRequired to false
serializationProxy =
new KeyedBackendSerializationProxy<>(
new ArtificialCNFExceptionThrowingClassLoader(
Thread.currentThread().getContextClassLoader(),
- cnfThrowingSerializerClasses));
+ cnfThrowingSerializerClasses),
+ false);
try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
serializationProxy.read(new DataInputViewStreamWrapper(in));
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
index c38f106..784b628 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
@@ -145,6 +145,8 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
protected abstract B getStateBackend() throws Exception;
+ protected abstract boolean isSerializerPresenceRequiredOnRestore();
+
protected CheckpointStreamFactory createStreamFactory() throws Exception {
if (checkpointStorageLocation == null) {
checkpointStorageLocation = getStateBackend()
@@ -1030,6 +1032,78 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
}
@Test
+ public void testSerializerPresenceOnRestore() throws Exception {
+ CheckpointStreamFactory streamFactory = createStreamFactory();
+ SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
+ Environment env = new DummyEnvironment();
+
+ AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE, env);
+
+ try {
+ ValueStateDescriptor<TestCustomStateClass> kvId = new ValueStateDescriptor<>("id", new TestReconfigurableCustomTypeSerializerPreUpgrade());
+ ValueState<TestCustomStateClass> state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
+
+ // ============== create snapshot, using the old serializer ==============
+
+ // make some modifications
+ backend.setCurrentKey(1);
+ state.update(new TestCustomStateClass("test-message-1", "this-should-be-ignored"));
+
+ backend.setCurrentKey(2);
+ state.update(new TestCustomStateClass("test-message-2", "this-should-be-ignored"));
+
+ KeyedStateHandle snapshot1 = runSnapshot(backend.snapshot(
+ 682375462378L,
+ 2,
+ streamFactory,
+ CheckpointOptions.forCheckpointWithDefaultLocation()));
+
+ snapshot1.registerSharedStates(sharedStateRegistry);
+ backend.dispose();
+
+ // ========== restore snapshot, using the new serializer (that has different classname) ==========
+
+ // on restore, simulate that the previous serializer class is no longer in the classloader
+ env = new DummyEnvironment(
+ new ArtificialCNFExceptionThrowingClassLoader(
+ getClass().getClassLoader(),
+ Collections.singleton(TestReconfigurableCustomTypeSerializerPreUpgrade.class.getName())));
+
+ try {
+ backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot1, env);
+ } catch (IOException e) {
+ if (!isSerializerPresenceRequiredOnRestore()) {
+ fail("Presence of old serializer should not have been required.");
+ } else {
+ // test success
+ return;
+ }
+ }
+
+ // if serializer presence is not required, continue on to modify some state to make sure that everything works correctly
+ kvId = new ValueStateDescriptor<>("id", new TestReconfigurableCustomTypeSerializerUpgraded());
+ state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
+
+ backend.setCurrentKey(1);
+ state.update(new TestCustomStateClass("new-test-message-1", "extra-message-1"));
+
+ backend.setCurrentKey(2);
+ state.update(new TestCustomStateClass("new-test-message-2", "extra-message-2"));
+
+ KeyedStateHandle snapshot2 = runSnapshot(backend.snapshot(
+ 682375462379L,
+ 3,
+ streamFactory,
+ CheckpointOptions.forCheckpointWithDefaultLocation()));
+
+ snapshot2.registerSharedStates(sharedStateRegistry);
+ snapshot1.discardState();
+ } finally {
+ backend.dispose();
+ }
+ }
+
+ @Test
@SuppressWarnings("unchecked")
public void testValueState() throws Exception {
CheckpointStreamFactory streamFactory = createStreamFactory();
@@ -4261,6 +4335,9 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
}
}
+ public static class TestReconfigurableCustomTypeSerializerPreUpgrade extends TestReconfigurableCustomTypeSerializer {}
+ public static class TestReconfigurableCustomTypeSerializerUpgraded extends TestReconfigurableCustomTypeSerializer {}
+
/**
* We throw this in our {@link ExceptionThrowingTestSerializer}.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 469247c..90d0fc6 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -588,8 +588,12 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
*/
private void restoreKVStateMetaData() throws IOException, StateMigrationException, RocksDBException {
+ // isSerializerPresenceRequired flag is set to false, since for the RocksDB state backend,
+ // deserialization of state happens lazily during runtime; we depend on the fact
+ // that the new serializer for states could be compatible, and therefore the restore can continue
+ // without old serializers required to be present.
KeyedBackendSerializationProxy<K> serializationProxy =
- new KeyedBackendSerializationProxy<>(rocksDBKeyedStateBackend.userCodeClassLoader);
+ new KeyedBackendSerializationProxy<>(rocksDBKeyedStateBackend.userCodeClassLoader, false);
serializationProxy.read(currentStateHandleInView);
@@ -925,8 +929,12 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
inputStream = metaStateHandle.openInputStream();
stateBackend.cancelStreamRegistry.registerCloseable(inputStream);
+ // isSerializerPresenceRequired flag is set to false, since for the RocksDB state backend,
+ // deserialization of state happens lazily during runtime; we depend on the fact
+ // that the new serializer for states could be compatible, and therefore the restore can continue
+ // without old serializers required to be present.
KeyedBackendSerializationProxy<T> serializationProxy =
- new KeyedBackendSerializationProxy<>(stateBackend.userCodeClassLoader);
+ new KeyedBackendSerializationProxy<>(stateBackend.userCodeClassLoader, false);
DataInputView in = new DataInputViewStreamWrapper(inputStream);
serializationProxy.read(in);
http://git-wip-us.apache.org/repos/asf/flink/blob/a7ed135f/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
index f54d0e3..ad89583 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
@@ -127,6 +127,11 @@ public class RocksDBStateBackendTest extends StateBackendTestBase<RocksDBStateBa
return backend;
}
+ @Override
+ protected boolean isSerializerPresenceRequiredOnRestore() {
+ return false;
+ }
+
// small safety net for instance cleanups, so that no native objects are left
@After
public void cleanupRocksDB() {