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/12/11 09:47:45 UTC
[flink] branch master updated (2ab4d41 -> 2f7bb1c)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2ab4d41 [FLINK-10743] [runtime] Use 0 processExitCode for ApplicationStatus.CANCELED
new bc41b75 [hotfix] [state backend, tests] Certain StateBackendMigrationTestBase tests should fail if exception isn't thrown
new 3b16b4e [hotfix] [state backends] New namespace serializer in HeapKeyedStateBackend should always be compatible
new 7fe1743 [hotfix] [state backends, tests] Make StateBackendMigrationTestBase more understandable
new b879a49 [FLINK-11094] [rocksdb] Eagerly create meta infos for restored state in RocksDBKeyedStateBackend
new ebc46c5 [FLINK-11094] [state backends] Let meta infos always lazily access restore serializer
new 97f556e [FLINK-11094] [state backends] State backends no longer need separate map for restored StateMetaInfoSnapshots
new 735e2f8 [hotfix] [tests] Remove unused enum from StateBackendTestBase
new 34ca8b7 [hotfix] Cleanup unused methods / appropriate method renames in StateMetaInfoSnapshot
new 8b898cd [FLINK-11087] [state] Incorrect broadcast state K/V serializer snapshot association when restoring from 1.5.x
new a30164b [FLINK-11087] [docs] Amend compatibility table to notify issue with restoring 1.5.x braodcast state in later versions.
new 7067798 [hotfix] [docs] Fix typo in Joining documentation
new 30a7c5a [hotfix] [docs] Fix tEnv in tableApi docs
new 0078a7c [hotfix] [docs] Correct the parameter in Operators Overview doc
new de454e5 [FLINK-11029] [docs] Fixed incorrect parameter in Working with state doc
new 2f7bb1c [FLINK-10359] [docs] Scala example in DataSet docs is broken
The 15 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/dev/batch/dataset_transformations.md | 4 +-
docs/dev/stream/operators/index.md | 4 +-
docs/dev/stream/operators/joining.md | 2 +-
docs/dev/stream/state/state.md | 2 +-
docs/dev/table/tableApi.md | 2 +-
docs/ops/upgrading.md | 5 +-
.../runtime/state/DefaultOperatorStateBackend.java | 58 +---
.../RegisteredBroadcastStateBackendMetaInfo.java | 84 ++++--
.../RegisteredKeyValueStateBackendMetaInfo.java | 115 +++++---
.../RegisteredOperatorStateBackendMetaInfo.java | 46 ++-
...egisteredPriorityQueueStateBackendMetaInfo.java | 39 ++-
.../runtime/state/RegisteredStateMetaInfoBase.java | 17 ++
.../runtime/state/StateSerializerProvider.java | 245 ++++++++++++++++
.../runtime/state/heap/HeapKeyedStateBackend.java | 78 ++---
.../state/metainfo/LegacyStateMetaInfoReaders.java | 33 ++-
.../state/metainfo/StateMetaInfoSnapshot.java | 33 +--
.../StateMetaInfoSnapshotReadersWriters.java | 2 +-
.../runtime/state/SerializationProxiesTest.java | 4 +-
.../state/StateBackendMigrationTestBase.java | 324 ++++-----------------
.../flink/runtime/state/StateBackendTestBase.java | 59 +++-
.../runtime/state/StateSerializerProviderTest.java | 187 ++++++++++++
.../runtime/testutils/statemigration/TestType.java | 237 +++++++++++++++
.../V1TestTypeSerializerSnapshot.java | 62 ++++
.../V2TestTypeSerializerSnapshot.java} | 35 +--
.../streaming/state/RocksDBKeyedStateBackend.java | 158 ++++------
.../StatefulJobWBroadcastStateMigrationITCase.java | 66 ++---
.../_metadata | Bin 20936 -> 21096 bytes
.../_metadata | Bin 20936 -> 21096 bytes
.../_metadata | Bin 22318 -> 22478 bytes
.../_metadata | Bin 42566 -> 42726 bytes
30 files changed, 1249 insertions(+), 652 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/state/StateSerializerProvider.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java
copy flink-runtime/src/{main/java/org/apache/flink/runtime/plugable/ReusingDeserializationDelegate.java => test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java} (51%)
[flink] 08/15: [hotfix] Cleanup unused methods / appropriate method
renames in StateMetaInfoSnapshot
Posted by tz...@apache.org.
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 34ca8b772f9ebb152210692d12295db78c534e17
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sun Dec 9 16:10:31 2018 +0800
[hotfix] Cleanup unused methods / appropriate method renames in StateMetaInfoSnapshot
This commit removes the `restoreTypeSerializer(...)` method.
That method is no longer used after the series of changes in
FLINK-11094. This also corresponds to the new principle that the restore
serializer is only accessed, when the state backends attempt request it
from their state meta infos. We do not create restore serializers
eagerly when creating meta infos from a StateMetaInfoSnapshot.
It also removes "config" from names of methods and fields
related to serializer snapshotting.
This corresponds to the abstraction rework of retiring
TypeSerializerConfigSnapshot to be replaced by TypeSerializerSnapshot.
The related fields / methods should not mention "config" anymore.
---
.../RegisteredBroadcastStateBackendMetaInfo.java | 4 +--
.../RegisteredKeyValueStateBackendMetaInfo.java | 4 +--
.../RegisteredOperatorStateBackendMetaInfo.java | 2 +-
...egisteredPriorityQueueStateBackendMetaInfo.java | 2 +-
.../state/metainfo/StateMetaInfoSnapshot.java | 33 ++++++++--------------
.../StateMetaInfoSnapshotReadersWriters.java | 2 +-
.../runtime/state/SerializationProxiesTest.java | 4 +--
7 files changed, 20 insertions(+), 31 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
index 95a650e..ecc13fa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
@@ -75,10 +75,10 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
snapshot.getOption(StateMetaInfoSnapshot.CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE)),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<K>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER))),
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER))),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<V>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.BROADCAST == snapshot.getBackendStateType());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
index ebe8e94..b37c79d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
@@ -87,10 +87,10 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
snapshot.getName(),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<N>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER))),
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER))),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<S>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
null);
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.KEY_VALUE == snapshot.getBackendStateType());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
index afb3d77..921947a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
@@ -73,7 +73,7 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
snapshot.getName(),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<S>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
OperatorStateHandle.Mode.valueOf(
snapshot.getOption(StateMetaInfoSnapshot.CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE)));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
index 60c88e3..961d96f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
@@ -51,7 +51,7 @@ public class RegisteredPriorityQueueStateBackendMetaInfo<T> extends RegisteredSt
snapshot.getName(),
StateSerializerProvider.fromRestoredState(
(TypeSerializerSnapshot<T>) Preconditions.checkNotNull(
- snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
+ snapshot.getTypeSerializerSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.PRIORITY_QUEUE == snapshot.getBackendStateType());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshot.java
index 1e9d919..9b05500 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshot.java
@@ -81,7 +81,7 @@ public class StateMetaInfoSnapshot {
/** The configurations of all the type serializers used with the state. */
@Nonnull
- private final Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshots;
+ private final Map<String, TypeSerializerSnapshot<?>> serializerSnapshots;
// TODO this will go away once all serializers have the restoreSerializer() factory method properly implemented.
/** The serializers used by the state. */
@@ -92,8 +92,8 @@ public class StateMetaInfoSnapshot {
@Nonnull String name,
@Nonnull BackendStateType backendStateType,
@Nonnull Map<String, String> options,
- @Nonnull Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshots) {
- this(name, backendStateType, options, serializerConfigSnapshots, new HashMap<>());
+ @Nonnull Map<String, TypeSerializerSnapshot<?>> serializerSnapshots) {
+ this(name, backendStateType, options, serializerSnapshots, new HashMap<>());
}
/**
@@ -106,12 +106,12 @@ public class StateMetaInfoSnapshot {
@Nonnull String name,
@Nonnull BackendStateType backendStateType,
@Nonnull Map<String, String> options,
- @Nonnull Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshots,
+ @Nonnull Map<String, TypeSerializerSnapshot<?>> serializerSnapshots,
@Nonnull Map<String, TypeSerializer<?>> serializers) {
this.name = name;
this.backendStateType = backendStateType;
this.options = options;
- this.serializerConfigSnapshots = serializerConfigSnapshots;
+ this.serializerSnapshots = serializerSnapshots;
this.serializers = serializers;
}
@@ -121,13 +121,13 @@ public class StateMetaInfoSnapshot {
}
@Nullable
- public TypeSerializerSnapshot<?> getTypeSerializerConfigSnapshot(@Nonnull String key) {
- return serializerConfigSnapshots.get(key);
+ public TypeSerializerSnapshot<?> getTypeSerializerSnapshot(@Nonnull String key) {
+ return serializerSnapshots.get(key);
}
@Nullable
- public TypeSerializerSnapshot<?> getTypeSerializerConfigSnapshot(@Nonnull CommonSerializerKeys key) {
- return getTypeSerializerConfigSnapshot(key.toString());
+ public TypeSerializerSnapshot<?> getTypeSerializerSnapshot(@Nonnull CommonSerializerKeys key) {
+ return getTypeSerializerSnapshot(key.toString());
}
@Nullable
@@ -150,20 +150,9 @@ public class StateMetaInfoSnapshot {
return name;
}
- @Nullable
- public TypeSerializer<?> restoreTypeSerializer(@Nonnull String key) {
- TypeSerializerSnapshot<?> configSnapshot = getTypeSerializerConfigSnapshot(key);
- return (configSnapshot != null) ? configSnapshot.restoreSerializer() : null;
- }
-
- @Nullable
- public TypeSerializer<?> restoreTypeSerializer(@Nonnull CommonSerializerKeys key) {
- return restoreTypeSerializer(key.toString());
- }
-
@Nonnull
- public Map<String, TypeSerializerSnapshot<?>> getSerializerConfigSnapshotsImmutable() {
- return Collections.unmodifiableMap(serializerConfigSnapshots);
+ public Map<String, TypeSerializerSnapshot<?>> getSerializerSnapshotsImmutable() {
+ return Collections.unmodifiableMap(serializerSnapshots);
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
index 4408dfc..ad1e7be 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
@@ -165,7 +165,7 @@ public class StateMetaInfoSnapshotReadersWriters {
@Nonnull DataOutputView outputView) throws IOException {
final Map<String, String> optionsMap = snapshot.getOptionsImmutable();
final Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshotsMap =
- snapshot.getSerializerConfigSnapshotsImmutable();
+ snapshot.getSerializerSnapshotsImmutable();
outputView.writeUTF(snapshot.getName());
outputView.writeInt(snapshot.getBackendStateType().ordinal());
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 c1f08e0..55aacb2 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
@@ -251,7 +251,7 @@ public class SerializationProxiesTest {
Assert.assertEquals(expected.getBackendStateType(), actual.getBackendStateType());
Assert.assertEquals(expected.getOptionsImmutable(), actual.getOptionsImmutable());
Assert.assertEquals(
- expected.getSerializerConfigSnapshotsImmutable(),
- actual.getSerializerConfigSnapshotsImmutable());
+ expected.getSerializerSnapshotsImmutable(),
+ actual.getSerializerSnapshotsImmutable());
}
}
[flink] 02/15: [hotfix] [state backends] New namespace serializer
in HeapKeyedStateBackend should always be compatible
Posted by tz...@apache.org.
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 3b16b4e454f20199bbe25cfeef9645e497e1e783
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Dec 7 20:47:30 2018 +0800
[hotfix] [state backends] New namespace serializer in HeapKeyedStateBackend should always be compatible
Previously, we were only checking if the new namespace serializer is
incompatible, while properly we should be checking that it is strictly
compatible.
This doesn't affect any user expected behaviour, since the namespace
serializer is never exposed to users.
---
.../org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 4eff3a2..ecad76c 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
@@ -283,8 +283,8 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
TypeSerializerSchemaCompatibility<N> namespaceCompatibility =
namespaceSerializerSnapshot.resolveSchemaCompatibility(namespaceSerializer);
- if (namespaceCompatibility.isIncompatible()) {
- throw new StateMigrationException("For heap backends, the new namespace serializer must not be incompatible.");
+ if (!namespaceCompatibility.isCompatibleAsIs()) {
+ throw new StateMigrationException("For heap backends, the new namespace serializer must be compatible.");
}
@SuppressWarnings("unchecked")
[flink] 06/15: [FLINK-11094] [state backends] State backends no
longer need separate map for restored StateMetaInfoSnapshots
Posted by tz...@apache.org.
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 97f556ebd122dfe16dd8e148d3ac7c5386fd08ac
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Dec 10 15:07:47 2018 +0800
[FLINK-11094] [state backends] State backends no longer need separate map for restored StateMetaInfoSnapshots
Since now all restored state meta info snapshots are handled so that we
always eagerly create the corresponding RegisteredStateMetaInfoBase for
it, the information is already part of the registered state infos map.
As can be seen in the changes, those maps are no longer queried and can
therefore be safely removed.
This closes #7264.
---
.../runtime/state/DefaultOperatorStateBackend.java | 58 +++--------------
.../runtime/state/heap/HeapKeyedStateBackend.java | 74 +++++-----------------
.../streaming/state/RocksDBKeyedStateBackend.java | 15 -----
3 files changed, 26 insertions(+), 121 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index 4702919..952dffb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -26,7 +26,6 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
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.api.common.typeutils.UnloadableDummyTypeSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.FSDataInputStream;
@@ -106,22 +105,9 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
private final boolean asynchronousSnapshots;
/**
- * Map of state names to their corresponding restored state meta info.
- *
- * <p>TODO this map can be removed when eager-state registration is in place.
- * TODO we currently need this cached to check state migration strategies when new serializers are registered.
- */
- private final Map<String, StateMetaInfoSnapshot> restoredOperatorStateMetaInfos;
-
- /**
- * Map of state names to their corresponding restored broadcast state meta info.
- */
- private final Map<String, StateMetaInfoSnapshot> restoredBroadcastStateMetaInfos;
-
- /**
* Cache of already accessed states.
*
- * <p>In contrast to {@link #registeredOperatorStates} and {@link #restoredOperatorStateMetaInfos} which may be repopulated
+ * <p>In contrast to {@link #registeredOperatorStates} which may be repopulated
* with restored state, this map is always empty at the beginning.
*
* <p>TODO this map should be moved to a base class once we have proper hierarchy for the operator state backends.
@@ -148,8 +134,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
this.asynchronousSnapshots = asynchronousSnapshots;
this.accessedStatesByName = new HashMap<>();
this.accessedBroadcastStatesByName = new HashMap<>();
- this.restoredOperatorStateMetaInfos = new HashMap<>();
- this.restoredBroadcastStateMetaInfos = new HashMap<>();
this.snapshotStrategy = new DefaultOperatorStateBackendSnapshotStrategy();
}
@@ -226,34 +210,22 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
broadcastState.getStateMetaInfo().getAssignmentMode(),
OperatorStateHandle.Mode.BROADCAST);
- final StateMetaInfoSnapshot metaInfoSnapshot = restoredBroadcastStateMetaInfos.get(name);
+ RegisteredBroadcastStateBackendMetaInfo<K, V> restoredBroadcastStateMetaInfo = broadcastState.getStateMetaInfo();
// check whether new serializers are incompatible
- TypeSerializerSnapshot<K> keySerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<K>) metaInfoSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER));
-
TypeSerializerSchemaCompatibility<K> keyCompatibility =
- keySerializerSnapshot.resolveSchemaCompatibility(broadcastStateKeySerializer);
+ restoredBroadcastStateMetaInfo.updateKeySerializer(broadcastStateKeySerializer);
if (keyCompatibility.isIncompatible()) {
throw new StateMigrationException("The new key serializer for broadcast state must not be incompatible.");
}
- TypeSerializerSnapshot<V> valueSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<V>) metaInfoSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
-
TypeSerializerSchemaCompatibility<V> valueCompatibility =
- valueSerializerSnapshot.resolveSchemaCompatibility(broadcastStateValueSerializer);
+ restoredBroadcastStateMetaInfo.updateValueSerializer(broadcastStateValueSerializer);
if (valueCompatibility.isIncompatible()) {
throw new StateMigrationException("The new value serializer for broadcast state must not be incompatible.");
}
- // new serializer is compatible; use it to replace the old serializer
- broadcastState.setStateMetaInfo(
- new RegisteredBroadcastStateBackendMetaInfo<>(
- name,
- OperatorStateHandle.Mode.BROADCAST,
- broadcastStateKeySerializer,
- broadcastStateValueSerializer));
+ broadcastState.setStateMetaInfo(restoredBroadcastStateMetaInfo);
}
accessedBroadcastStatesByName.put(name, broadcastState);
@@ -345,8 +317,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
" not be loaded. This is a temporary restriction that will be fixed in future versions.");
}
- restoredOperatorStateMetaInfos.put(restoredSnapshot.getName(), restoredSnapshot);
-
PartitionableListState<?> listState = registeredOperatorStates.get(restoredSnapshot.getName());
if (null == listState) {
@@ -381,8 +351,6 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
" not be loaded. This is a temporary restriction that will be fixed in future versions.");
}
- restoredBroadcastStateMetaInfos.put(restoredSnapshot.getName(), restoredSnapshot);
-
BackendWritableBroadcastState<? ,?> broadcastState = registeredBroadcastStates.get(restoredSnapshot.getName());
if (broadcastState == null) {
@@ -590,25 +558,19 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
partitionableListState.getStateMetaInfo().getAssignmentMode(),
mode);
- StateMetaInfoSnapshot restoredSnapshot = restoredOperatorStateMetaInfos.get(name);
- RegisteredOperatorStateBackendMetaInfo<S> metaInfo =
- new RegisteredOperatorStateBackendMetaInfo<>(restoredSnapshot);
+ RegisteredOperatorStateBackendMetaInfo<S> restoredPartitionableListStateMetaInfo =
+ partitionableListState.getStateMetaInfo();
- // check compatibility to determine if state migration is required
+ // check compatibility to determine if new serializers are incompatible
TypeSerializer<S> newPartitionStateSerializer = partitionStateSerializer.duplicate();
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<S> stateSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<S>) restoredSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
-
TypeSerializerSchemaCompatibility<S> stateCompatibility =
- stateSerializerSnapshot.resolveSchemaCompatibility(newPartitionStateSerializer);
+ restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);
if (stateCompatibility.isIncompatible()) {
throw new StateMigrationException("The new state serializer for operator state must not be incompatible.");
}
- partitionableListState.setStateMetaInfo(
- new RegisteredOperatorStateBackendMetaInfo<>(name, newPartitionStateSerializer, mode));
+ partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);
}
accessedStatesByName.put(name, partitionableListState);
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 ecad76c..3f8761b 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
@@ -30,7 +30,6 @@ import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
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.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
@@ -79,6 +78,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.IOException;
import java.io.InputStream;
@@ -126,14 +126,6 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
private final Map<String, HeapPriorityQueueSnapshotRestoreWrapper> registeredPQStates;
/**
- * Map of state names to their corresponding restored state meta info.
- *
- * <p>TODO this map can be removed when eager-state registration is in place.
- * TODO we currently need this cached to check state migration strategies when new serializers are registered.
- */
- private final Map<StateUID, StateMetaInfoSnapshot> restoredStateMetaInfo;
-
- /**
* The configuration for local recovery.
*/
private final LocalRecoveryConfig localRecoveryConfig;
@@ -173,7 +165,6 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
this.snapshotStrategy = new HeapSnapshotStrategy(synchronicityTrait);
LOG.info("Initializing heap keyed state backend with stream factory.");
- this.restoredStateMetaInfo = new HashMap<>();
this.priorityQueueSetFactory = priorityQueueSetFactory;
}
@@ -194,23 +185,9 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
// TODO we implement the simple way of supporting the current functionality, mimicking keyed state
// because this should be reworked in FLINK-9376 and then we should have a common algorithm over
// StateMetaInfoSnapshot that avoids this code duplication.
- StateMetaInfoSnapshot restoredMetaInfoSnapshot =
- restoredStateMetaInfo.get(StateUID.of(stateName, StateMetaInfoSnapshot.BackendStateType.PRIORITY_QUEUE));
-
- Preconditions.checkState(
- restoredMetaInfoSnapshot != null,
- "Requested to check compatibility of a restored RegisteredKeyedBackendStateMetaInfo," +
- " but its corresponding restored snapshot cannot be found.");
-
- StateMetaInfoSnapshot.CommonSerializerKeys serializerKey =
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER;
-
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<T> serializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<T>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(serializerKey));
TypeSerializerSchemaCompatibility<T> compatibilityResult =
- serializerSnapshot.resolveSchemaCompatibility(byteOrderedElementSerializer);
+ existingState.getMetaInfo().updateElementSerializer(byteOrderedElementSerializer);
if (compatibilityResult.isIncompatible()) {
throw new FlinkRuntimeException(new StateMigrationException("For heap backends, the new priority queue serializer must not be incompatible."));
@@ -252,57 +229,42 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
private <N, V> StateTable<K, N, V> tryRegisterStateTable(
TypeSerializer<N> namespaceSerializer,
StateDescriptor<?, V> stateDesc,
- StateSnapshotTransformer<V> snapshotTransformer) throws StateMigrationException {
+ @Nullable StateSnapshotTransformer<V> snapshotTransformer) throws StateMigrationException {
@SuppressWarnings("unchecked")
StateTable<K, N, V> stateTable = (StateTable<K, N, V>) registeredKVStates.get(stateDesc.getName());
TypeSerializer<V> newStateSerializer = stateDesc.getSerializer();
- RegisteredKeyValueStateBackendMetaInfo<N, V> newMetaInfo = new RegisteredKeyValueStateBackendMetaInfo<>(
- stateDesc.getType(),
- stateDesc.getName(),
- namespaceSerializer,
- newStateSerializer,
- snapshotTransformer);
if (stateTable != null) {
- @SuppressWarnings("unchecked")
- StateMetaInfoSnapshot restoredMetaInfoSnapshot =
- restoredStateMetaInfo.get(
- StateUID.of(stateDesc.getName(), StateMetaInfoSnapshot.BackendStateType.KEY_VALUE));
-
- Preconditions.checkState(
- restoredMetaInfoSnapshot != null,
- "Requested to check compatibility of a restored RegisteredKeyedBackendStateMetaInfo," +
- " but its corresponding restored snapshot cannot be found.");
+ RegisteredKeyValueStateBackendMetaInfo<N, V> restoredKvMetaInfo = stateTable.getMetaInfo();
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<N> namespaceSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<N>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
- StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString()));
+ restoredKvMetaInfo.updateSnapshotTransformer(snapshotTransformer);
TypeSerializerSchemaCompatibility<N> namespaceCompatibility =
- namespaceSerializerSnapshot.resolveSchemaCompatibility(namespaceSerializer);
+ restoredKvMetaInfo.updateNamespaceSerializer(namespaceSerializer);
if (!namespaceCompatibility.isCompatibleAsIs()) {
throw new StateMigrationException("For heap backends, the new namespace serializer must be compatible.");
}
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<V> stateSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<V>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString()));
-
- RegisteredKeyValueStateBackendMetaInfo.checkStateMetaInfo(restoredMetaInfoSnapshot, stateDesc);
+ restoredKvMetaInfo.checkStateMetaInfo(stateDesc);
TypeSerializerSchemaCompatibility<V> stateCompatibility =
- stateSerializerSnapshot.resolveSchemaCompatibility(newStateSerializer);
+ restoredKvMetaInfo.updateStateSerializer(newStateSerializer);
if (stateCompatibility.isIncompatible()) {
throw new StateMigrationException("For heap backends, the new state serializer must not be incompatible.");
}
- stateTable.setMetaInfo(newMetaInfo);
+ stateTable.setMetaInfo(restoredKvMetaInfo);
} else {
+ RegisteredKeyValueStateBackendMetaInfo<N, V> newMetaInfo = new RegisteredKeyValueStateBackendMetaInfo<>(
+ stateDesc.getType(),
+ stateDesc.getName(),
+ namespaceSerializer,
+ newStateSerializer,
+ snapshotTransformer);
+
stateTable = snapshotStrategy.newStateTable(newMetaInfo);
registeredKVStates.put(stateDesc.getName(), stateTable);
}
@@ -536,10 +498,6 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
Map<Integer, StateMetaInfoSnapshot> kvStatesById) {
for (StateMetaInfoSnapshot metaInfoSnapshot : restoredMetaInfo) {
- restoredStateMetaInfo.put(
- StateUID.of(metaInfoSnapshot.getName(), metaInfoSnapshot.getBackendStateType()),
- metaInfoSnapshot);
-
final StateSnapshotRestore registeredState;
switch (metaInfoSnapshot.getBackendStateType()) {
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 065213b..a37f8aa 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
@@ -111,7 +111,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -212,14 +211,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
*/
private final LinkedHashMap<String, Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase>> kvStateInformation;
- /**
- * Map of state names to their corresponding restored state meta info.
- *
- * <p>TODO this map can be removed when eager-state registration is in place.
- * TODO we currently need this cached to check state migration strategies when new serializers are registered.
- */
- private final Map<String, StateMetaInfoSnapshot> restoredKvStateMetaInfos;
-
/** Number of bytes required to prefix the key groups. */
private final int keyGroupPrefixBytes;
@@ -296,7 +287,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
this.keyGroupPrefixBytes =
RocksDBKeySerializationUtils.computeRequiredBytesInKeyGroupPrefix(getNumberOfKeyGroups());
this.kvStateInformation = new LinkedHashMap<>();
- this.restoredKvStateMetaInfos = new HashMap<>();
this.writeOptions = new WriteOptions().setDisableWAL(true);
@@ -424,7 +414,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
IOUtils.closeQuietly(dbOptions);
IOUtils.closeQuietly(writeOptions);
kvStateInformation.clear();
- restoredKvStateMetaInfos.clear();
cleanInstanceBasePath();
}
@@ -510,7 +499,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
// clear all meta data
kvStateInformation.clear();
- restoredKvStateMetaInfos.clear();
try {
RocksDBIncrementalRestoreOperation<K> incrementalRestoreOperation = null;
@@ -753,8 +741,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
nameBytes,
rocksDBKeyedStateBackend.columnOptions);
- rocksDBKeyedStateBackend.restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);
-
ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);
// create a meta info for the state on restore;
@@ -1166,7 +1152,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
stateBackend.columnOptions);
columnFamilyDescriptors.add(columnFamilyDescriptor);
- stateBackend.restoredKvStateMetaInfos.put(stateMetaInfoSnapshot.getName(), stateMetaInfoSnapshot);
}
return columnFamilyDescriptors;
}
[flink] 10/15: [FLINK-11087] [docs] Amend compatibility table to
notify issue with restoring 1.5.x braodcast state in later versions.
Posted by tz...@apache.org.
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 a30164bf06f5d955771025e0341bccf6750f4062
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Dec 7 00:34:46 2018 +0800
[FLINK-11087] [docs] Amend compatibility table to notify issue with restoring 1.5.x braodcast state in later versions.
This closes #7256.
---
docs/ops/upgrading.md | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/docs/ops/upgrading.md b/docs/ops/upgrading.md
index 0451940..22cea4c 100644
--- a/docs/ops/upgrading.md
+++ b/docs/ops/upgrading.md
@@ -274,7 +274,10 @@ Savepoints are compatible across Flink versions as indicated by the table below:
<td class="text-center">O</td>
<td class="text-center">O</td>
<td class="text-center">O</td>
- <td class="text-left"></td>
+ <td class="text-left">There is a known issue with resuming broadcast state created with 1.5.x in versions
+ 1.6.x up to 1.6.2, and 1.7.0: <a href="https://issues.apache.org/jira/browse/FLINK-11087">FLINK-11087</a>. Users
+ upgrading to 1.6.x or 1.7.x series need to directly migrate to minor versions higher than 1.6.2 and 1.7.0,
+ respectively.</td>
</tr>
<tr>
<td class="text-center"><strong>1.6.x</strong></td>
[flink] 09/15: [FLINK-11087] [state] Incorrect broadcast state K/V
serializer snapshot association when restoring from 1.5.x
Posted by tz...@apache.org.
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 8b898cd342faa9efaa14f436bedaee11e6389252
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Dec 7 00:14:13 2018 +0800
[FLINK-11087] [state] Incorrect broadcast state K/V serializer snapshot association when restoring from 1.5.x
When restoring a broadcast state's meta information from a 1.5.x
savepoint, the
LegacyStateMetaInfoReaders.OperatorBackendStateMetaInfoReaderV2V3
incorrectly associates the first restored serializer as the value
serializer, and the second restored serializer as the key serializer.
The actual order of this should be the other way around.
This bug prevents successful broadcast state restores from 1.5, both for
Flink 1.6.x and 1.7.0.
The commit also modifies the StatefulJobWBroadcastStateMigrationITCase
to have different key / value types for its tested broadcast tests,
which otherwise would not have caught this bug.
---
.../state/metainfo/LegacyStateMetaInfoReaders.java | 33 ++++++-----
.../StatefulJobWBroadcastStateMigrationITCase.java | 66 ++++++++++-----------
.../_metadata | Bin 20936 -> 21096 bytes
.../_metadata | Bin 20936 -> 21096 bytes
.../_metadata | Bin 22318 -> 22478 bytes
.../_metadata | Bin 42566 -> 42726 bytes
6 files changed, 51 insertions(+), 48 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
index 77c267a..836edef 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
@@ -132,11 +132,6 @@ public class LegacyStateMetaInfoReaders {
static final OperatorBackendStateMetaInfoReaderV2V3 INSTANCE = new OperatorBackendStateMetaInfoReaderV2V3();
- private static final String[] ORDERED_KEY_STRINGS =
- new String[]{
- StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER.toString(),
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString()};
-
@Nonnull
@Override
public StateMetaInfoSnapshot readStateMetaInfoSnapshot(
@@ -156,17 +151,25 @@ public class LegacyStateMetaInfoReaders {
final int listSize = stateSerializerAndConfigList.size();
StateMetaInfoSnapshot.BackendStateType stateType = listSize == 1 ?
StateMetaInfoSnapshot.BackendStateType.OPERATOR : StateMetaInfoSnapshot.BackendStateType.BROADCAST;
- Map<String, TypeSerializerSnapshot<?>> serializerConfigsMap = new HashMap<>(listSize);
- for (int i = 0; i < listSize; ++i) {
- Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>> serializerAndConf =
- stateSerializerAndConfigList.get(i);
-
- // this particular mapping happens to support both, V2 and V3
- String serializerKey = ORDERED_KEY_STRINGS[ORDERED_KEY_STRINGS.length - 1 - i];
- serializerConfigsMap.put(
- serializerKey,
- serializerAndConf.f1);
+ Map<String, TypeSerializerSnapshot<?>> serializerConfigsMap = new HashMap<>(listSize);
+ switch (stateType) {
+ case OPERATOR:
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(0).f1);
+ break;
+ case BROADCAST:
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(0).f1);
+
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(1).f1);
+ break;
+ default:
+ throw new IllegalStateException("Unknown operator state type " + stateType);
}
return new StateMetaInfoSnapshot(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
index f792362..3f49d84 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
@@ -113,17 +113,17 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
expectedFirstState.put(2L, 2L);
expectedFirstState.put(3L, 3L);
- final Map<String, String> expectedSecondState = new HashMap<>();
- expectedSecondState.put("0", "0");
- expectedSecondState.put("1", "1");
- expectedSecondState.put("2", "2");
- expectedSecondState.put("3", "3");
-
- final Map<String, String> expectedThirdState = new HashMap<>();
- expectedThirdState.put("0", "0");
- expectedThirdState.put("1", "1");
- expectedThirdState.put("2", "2");
- expectedThirdState.put("3", "3");
+ final Map<String, Long> expectedSecondState = new HashMap<>();
+ expectedSecondState.put("0", 0L);
+ expectedSecondState.put("1", 1L);
+ expectedSecondState.put("2", 2L);
+ expectedSecondState.put("3", 3L);
+
+ final Map<Long, String> expectedThirdState = new HashMap<>();
+ expectedThirdState.put(0L, "0");
+ expectedThirdState.put(1L, "1");
+ expectedThirdState.put(2L, "2");
+ expectedThirdState.put(3L, "3");
if (executionMode == StatefulJobSavepointMigrationITCase.ExecutionMode.PERFORM_SAVEPOINT) {
nonParallelSource = new MigrationTestUtils.CheckpointingNonParallelSourceWithListState(NUM_SOURCE_ELEMENTS);
@@ -171,12 +171,12 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
"broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
- final MapStateDescriptor<String, String> secondBroadcastStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ final MapStateDescriptor<String, Long> secondBroadcastStateDesc = new MapStateDescriptor<>(
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
- final MapStateDescriptor<String, String> thirdBroadcastStateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ final MapStateDescriptor<Long, String> thirdBroadcastStateDesc = new MapStateDescriptor<>(
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
BroadcastStream<Tuple2<Long, Long>> npBroadcastStream = env
@@ -234,7 +234,7 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
private MapStateDescriptor<Long, Long> firstStateDesc;
- private MapStateDescriptor<String, String> secondStateDesc;
+ private MapStateDescriptor<String, Long> secondStateDesc;
@Override
public void open(Configuration parameters) throws Exception {
@@ -245,7 +245,7 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
);
secondStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
}
@@ -257,7 +257,7 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
@Override
public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
ctx.getBroadcastState(firstStateDesc).put(value.f0, value.f1);
- ctx.getBroadcastState(secondStateDesc).put(Long.toString(value.f0), Long.toString(value.f1));
+ ctx.getBroadcastState(secondStateDesc).put(Long.toString(value.f0), value.f1);
}
}
@@ -269,14 +269,14 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
private static final long serialVersionUID = 1333992081671604521L;
- private MapStateDescriptor<String, String> stateDesc;
+ private MapStateDescriptor<Long, String> stateDesc;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
stateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
}
@@ -287,7 +287,7 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
@Override
public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
- ctx.getBroadcastState(stateDesc).put(Long.toString(value.f0), Long.toString(value.f1));
+ ctx.getBroadcastState(stateDesc).put(value.f0, Long.toString(value.f1));
}
}
@@ -301,13 +301,13 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
private final Map<Long, Long> expectedFirstState;
- private final Map<String, String> expectedSecondState;
+ private final Map<String, Long> expectedSecondState;
private MapStateDescriptor<Long, Long> firstStateDesc;
- private MapStateDescriptor<String, String> secondStateDesc;
+ private MapStateDescriptor<String, Long> secondStateDesc;
- CheckingKeyedBroadcastFunction(Map<Long, Long> firstState, Map<String, String> secondState) {
+ CheckingKeyedBroadcastFunction(Map<Long, Long> firstState, Map<String, Long> secondState) {
this.expectedFirstState = firstState;
this.expectedSecondState = secondState;
}
@@ -321,7 +321,7 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
);
secondStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
}
@@ -334,8 +334,8 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
}
Assert.assertEquals(expectedFirstState, actualFirstState);
- final Map<String, String> actualSecondState = new HashMap<>();
- for (Map.Entry<String, String> entry: ctx.getBroadcastState(secondStateDesc).immutableEntries()) {
+ final Map<String, Long> actualSecondState = new HashMap<>();
+ for (Map.Entry<String, Long> entry: ctx.getBroadcastState(secondStateDesc).immutableEntries()) {
actualSecondState.put(entry.getKey(), entry.getValue());
}
Assert.assertEquals(expectedSecondState, actualSecondState);
@@ -357,11 +357,11 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
private static final long serialVersionUID = 1333992081671604521L;
- private final Map<String, String> expectedState;
+ private final Map<Long, String> expectedState;
- private MapStateDescriptor<String, String> stateDesc;
+ private MapStateDescriptor<Long, String> stateDesc;
- CheckingKeyedSingleBroadcastFunction(Map<String, String> state) {
+ CheckingKeyedSingleBroadcastFunction(Map<Long, String> state) {
this.expectedState = state;
}
@@ -370,14 +370,14 @@ public class StatefulJobWBroadcastStateMigrationITCase extends SavepointMigratio
super.open(parameters);
stateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
}
@Override
public void processElement(Tuple2<Long, Long> value, ReadOnlyContext ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
- final Map<String, String> actualState = new HashMap<>();
- for (Map.Entry<String, String> entry: ctx.getBroadcastState(stateDesc).immutableEntries()) {
+ final Map<Long, String> actualState = new HashMap<>();
+ for (Map.Entry<Long, String> entry: ctx.getBroadcastState(stateDesc).immutableEntries()) {
actualState.put(entry.getKey(), entry.getValue());
}
Assert.assertEquals(expectedState, actualState);
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata
index aed6183..2e02f1b 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata
index 13e8d32..4ca91d3 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata
index 9e2b086..8b2a63a 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata
index cceb5bf..cde6d5c 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata differ
[flink] 04/15: [FLINK-11094] [rocksdb] Eagerly create meta infos
for restored state in RocksDBKeyedStateBackend
Posted by tz...@apache.org.
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 b879a490ec0ab3114bf68170302aded3e9e7ad3d
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sun Dec 9 15:51:59 2018 +0800
[FLINK-11094] [rocksdb] Eagerly create meta infos for restored state in RocksDBKeyedStateBackend
This ensures that all restored state, even non-accessed ones after the
restore, have a meta info available on future snapshots.
---
.../runtime/state/RegisteredStateMetaInfoBase.java | 17 +++++++++++++++++
.../streaming/state/RocksDBKeyedStateBackend.java | 18 +++++++++++++++---
2 files changed, 32 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
index 4132d14..b7dff59 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredStateMetaInfoBase.java
@@ -42,4 +42,21 @@ public abstract class RegisteredStateMetaInfoBase {
@Nonnull
public abstract StateMetaInfoSnapshot snapshot();
+
+ public static RegisteredStateMetaInfoBase fromMetaInfoSnapshot(@Nonnull StateMetaInfoSnapshot snapshot) {
+
+ final StateMetaInfoSnapshot.BackendStateType backendStateType = snapshot.getBackendStateType();
+ switch (backendStateType) {
+ case KEY_VALUE:
+ return new RegisteredKeyValueStateBackendMetaInfo<>(snapshot);
+ case OPERATOR:
+ return new RegisteredOperatorStateBackendMetaInfo<>(snapshot);
+ case BROADCAST:
+ return new RegisteredBroadcastStateBackendMetaInfo<>(snapshot);
+ case PRIORITY_QUEUE:
+ return new RegisteredPriorityQueueStateBackendMetaInfo<>(snapshot);
+ default:
+ throw new IllegalArgumentException("Unknown backend state type: " + backendStateType);
+ }
+ }
}
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 50caa0d..d12370b 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
@@ -758,7 +758,11 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);
- registeredColumn = new Tuple2<>(columnFamily, null);
+ // create a meta info for the state on restore;
+ // this allows us to retain the state in future snapshots even if it wasn't accessed
+ RegisteredStateMetaInfoBase stateMetaInfo =
+ RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(restoredMetaInfo);
+ registeredColumn = new Tuple2<>(columnFamily, stateMetaInfo);
rocksDBKeyedStateBackend.kvStateInformation.put(restoredMetaInfo.getName(), registeredColumn);
} else {
@@ -1069,10 +1073,14 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
stateBackend.kvStateInformation.get(stateMetaInfoSnapshot.getName());
if (null == registeredStateMetaInfoEntry) {
+ // create a meta info for the state on restore;
+ // this allows us to retain the state in future snapshots even if it wasn't accessed
+ RegisteredStateMetaInfoBase stateMetaInfo =
+ RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
registeredStateMetaInfoEntry =
new Tuple2<>(
columnFamilyHandle != null ? columnFamilyHandle : stateBackend.db.createColumnFamily(columnFamilyDescriptor),
- null);
+ stateMetaInfo);
stateBackend.registerKvStateInformation(
stateMetaInfoSnapshot.getName(),
@@ -1201,9 +1209,13 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
ColumnFamilyHandle columnFamilyHandle = columnFamilyHandles.get(i);
+ // create a meta info for the state on restore;
+ // this allows us to retain the state in future snapshots even if it wasn't accessed
+ RegisteredStateMetaInfoBase stateMetaInfo =
+ RegisteredStateMetaInfoBase.fromMetaInfoSnapshot(stateMetaInfoSnapshot);
stateBackend.registerKvStateInformation(
stateMetaInfoSnapshot.getName(),
- new Tuple2<>(columnFamilyHandle, null));
+ new Tuple2<>(columnFamilyHandle, stateMetaInfo));
}
// use the restore sst files as the base for succeeding checkpoints
[flink] 12/15: [hotfix] [docs] Fix tEnv in tableApi docs
Posted by tz...@apache.org.
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 30a7c5a978e4ef30e289986f5b2e8107af2ba245
Author: unknown <ma...@ad1-zwt.corp.qihoo.net>
AuthorDate: Thu Dec 6 21:38:42 2018 +0800
[hotfix] [docs] Fix tEnv in tableApi docs
This closes #7254.
---
docs/dev/table/tableApi.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index 80930ca..f308aca 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -59,7 +59,7 @@ Table counts = orders
.select("a, b.count as cnt");
// conversion to DataSet
-DataSet<Row> result = tableEnv.toDataSet(counts, Row.class);
+DataSet<Row> result = tEnv.toDataSet(counts, Row.class);
result.print();
{% endhighlight %}
[flink] 01/15: [hotfix] [state backend,
tests] Certain StateBackendMigrationTestBase tests should fail if
exception isn't thrown
Posted by tz...@apache.org.
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 bc41b756ff180d9903ba22c0d83635c38a11098f
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Dec 7 20:42:52 2018 +0800
[hotfix] [state backend, tests] Certain StateBackendMigrationTestBase tests should fail if exception isn't thrown
This commit strengthens tests in StateBackendMigrationTestBase that
depend on a certain state operation (restoring state, accessing state,
etc.) to be failing to assert correct behaviour. However, we previously
do not really fail the test if no exception was thrown when there should
be.
This also caught some bugs in the test itself which had the tests
verifying incorrect behaviour.
---
.../state/StateBackendMigrationTestBase.java | 27 ++++++++++++++++++----
1 file changed, 23 insertions(+), 4 deletions(-)
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 f5f30d5..dff49f2 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
@@ -245,6 +245,8 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
}finally {
@@ -288,10 +290,12 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
kvId = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
} finally {
@@ -358,6 +362,8 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
// the new key serializer is incompatible; this should fail the restore
restoreKeyedBackend(new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE), snapshot);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
}
@@ -365,6 +371,8 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
// the new key serializer requires migration; this should fail the restore
restoreKeyedBackend(new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION), snapshot);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
}
@@ -397,26 +405,33 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
KeyedStateHandle snapshot = runSnapshot(
backend.snapshot(1L, 2L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
sharedStateRegistry);
- backend.dispose();
+ // test incompatible namespace serializer; start with a freshly restored backend
+ backend.dispose();
backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot);
-
try {
// the new namespace serializer is incompatible; this should fail the restore
backend.getPartitionedState(
new TestType("namespace", 123),
new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE),
kvId);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
}
+ // test namespace serializer that requires migration; start with a freshly restored backend
+ backend.dispose();
+ backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot);
try {
// the new namespace serializer requires migration; this should fail the restore
backend.getPartitionedState(
new TestType("namespace", 123),
new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION),
kvId);
+
+ Assert.fail("should have failed");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
}
@@ -685,10 +700,12 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new MapStateDescriptor<>(
stateName,
IntSerializer.INSTANCE,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
// the new value serializer is INCOMPATIBLE, so registering the state should fail
backend.getBroadcastState(descriptor);
+
+ Assert.fail("should have failed.");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
} finally {
@@ -726,6 +743,8 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// the new key serializer is INCOMPATIBLE, so registering the state should fail
backend.getBroadcastState(descriptor);
+
+ Assert.fail("should have failed.");
} catch (Exception e) {
Assert.assertTrue(ExceptionUtils.findThrowable(e, StateMigrationException.class).isPresent());
} finally {
[flink] 15/15: [FLINK-10359] [docs] Scala example in DataSet docs
is broken
Posted by tz...@apache.org.
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 2f7bb1c89edc99fe6b433bec3000cdf087c2cf9e
Author: yanghua <ya...@gmail.com>
AuthorDate: Mon Dec 10 11:00:29 2018 +0800
[FLINK-10359] [docs] Scala example in DataSet docs is broken
This closes #7266.
---
docs/dev/batch/dataset_transformations.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/dev/batch/dataset_transformations.md b/docs/dev/batch/dataset_transformations.md
index 96c04c9..cf2da3f 100644
--- a/docs/dev/batch/dataset_transformations.md
+++ b/docs/dev/batch/dataset_transformations.md
@@ -712,7 +712,7 @@ class MyCombinableGroupReducer
out: Collector[String]): Unit =
{
val r: (String, Int) =
- in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
+ in.iterator.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
// concat key and sum and emit
out.collect (r._1 + "-" + r._2)
}
@@ -722,7 +722,7 @@ class MyCombinableGroupReducer
out: Collector[(String, Int)]): Unit =
{
val r: (String, Int) =
- in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
+ in.iterator.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
// emit tuple with key and sum
out.collect(r)
}
[flink] 11/15: [hotfix] [docs] Fix typo in Joining documentation
Posted by tz...@apache.org.
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 7067798c25e2152eed6ffc11639da2b9f5064d4e
Author: zhangzhanchun <zh...@didichuxing.com>
AuthorDate: Tue Dec 11 11:00:49 2018 +0800
[hotfix] [docs] Fix typo in Joining documentation
This closes #7270.
---
docs/dev/stream/operators/joining.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/dev/stream/operators/joining.md b/docs/dev/stream/operators/joining.md
index b95aadd..7e9915e 100644
--- a/docs/dev/stream/operators/joining.md
+++ b/docs/dev/stream/operators/joining.md
@@ -101,7 +101,7 @@ orangeStream.join(greenStream)
</div>
## Sliding Window Join
-When performing a sliding window join, all elements with a common key and common sliding window are joined are pairwise combinations and passed on to the `JoinFunction` or `FlatJoinFunction`. Elements of one stream that do not have elements from the other stream in the current sliding window are not emitted! Note that some elements might be joined in one sliding window but not in another!
+When performing a sliding window join, all elements with a common key and common sliding window are joined as pairwise combinations and passed on to the `JoinFunction` or `FlatJoinFunction`. Elements of one stream that do not have elements from the other stream in the current sliding window are not emitted! Note that some elements might be joined in one sliding window but not in another!
<img src="{{ site.baseurl }}/fig/sliding-window-join.svg" class="center" style="width: 80%;" />
[flink] 14/15: [FLINK-11029] [docs] Fixed incorrect parameter in
Working with state doc
Posted by tz...@apache.org.
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 de454e5c62fe333b5f9ecc84f3b86c334e4026ba
Author: KarmaGYZ <ka...@gmail.com>
AuthorDate: Thu Nov 29 21:02:54 2018 +0800
[FLINK-11029] [docs] Fixed incorrect parameter in Working with state doc
This closes #7198.
---
docs/dev/stream/state/state.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/dev/stream/state/state.md b/docs/dev/stream/state/state.md
index decf1db..a3109f2 100644
--- a/docs/dev/stream/state/state.md
+++ b/docs/dev/stream/state/state.md
@@ -142,7 +142,7 @@ is available in a `RichFunction` has these methods for accessing state:
* `ValueState<T> getState(ValueStateDescriptor<T>)`
* `ReducingState<T> getReducingState(ReducingStateDescriptor<T>)`
* `ListState<T> getListState(ListStateDescriptor<T>)`
-* `AggregatingState<IN, OUT> getAggregatingState(AggregatingState<IN, OUT>)`
+* `AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)`
* `FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)`
* `MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)`
[flink] 03/15: [hotfix] [state backends,
tests] Make StateBackendMigrationTestBase more understandable
Posted by tz...@apache.org.
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 7fe1743c6ce60b2587db2d7992c75738070f8dfd
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sat Dec 8 18:28:13 2018 +0800
[hotfix] [state backends, tests] Make StateBackendMigrationTestBase more understandable
The StateBackendMigrationTestBase previously mocked user behaviour of
upgrading serializers by using a single serializer class, that can be
configured with different target compatibility results when they are
checked for compatibility.
This is a bit hard to understand, also doesn't really reflect how a user
would actually approach the feature. Instead, instead of configuring a
single serializer class with different compatibility "personalities",
this commit uses actual different classes, V1TestTypeSerializer,
V2TestTypeSerializer, and IncompatibleTestTypeSerializer, to simulate
the compatibility cases in tests.
This commit also refactors the serializer migration related test
serializers / types / snapshots to the testutil package, so that it can
be shared by other state migration related tests in the future.
---
.../state/StateBackendMigrationTestBase.java | 301 +++------------------
.../runtime/testutils/statemigration/TestType.java | 237 ++++++++++++++++
.../V1TestTypeSerializerSnapshot.java | 62 +++++
.../V2TestTypeSerializerSnapshot.java | 60 ++++
4 files changed, 395 insertions(+), 265 deletions(-)
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 dff49f2..5511792 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
@@ -34,21 +34,20 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
-import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+import org.apache.flink.runtime.testutils.statemigration.TestType;
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.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
-import java.util.Objects;
import java.util.concurrent.RunnableFuture;
/**
@@ -69,17 +68,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// lazily initialized stream storage
private CheckpointStorageLocation checkpointStorageLocation;
- /**
- * The compatibility behaviour of {@link TestSerializer}.
- * This controls what format the serializer writes in, as well as
- * the result of the compatibility check against the prior serializer snapshot.
- */
- public enum SerializerCompatibilityType {
- COMPATIBLE_AS_IS,
- REQUIRES_MIGRATION,
- INCOMPATIBLE
- }
-
// -------------------------------------------------------------------------------
// Keyed state backend migration tests
// -------------------------------------------------------------------------------
@@ -95,7 +83,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ValueStateDescriptor<TestType> kvId = new ValueStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ValueState<TestType> valueState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -113,10 +101,10 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot);
- // the new serializer is REQUIRES_MIGRATION, and has a completely new serialization schema.
+ // the new serializer is V2, and has a completely new serialization schema.
kvId = new ValueStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestType.V2TestTypeSerializer());
valueState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -151,7 +139,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> kvId = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> listState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -174,10 +162,10 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot);
- // the new serializer is REQUIRES_MIGRATION, and has a completely new serialization schema.
+ // the new serializer is V2, and has a completely new serialization schema.
kvId = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestType.V2TestTypeSerializer());
listState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -221,7 +209,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ValueStateDescriptor<TestType> kvId = new ValueStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ValueState<TestType> valueState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -241,7 +229,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
kvId = new ValueStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ new TestType.IncompatibleTestTypeSerializer());
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -265,7 +253,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> kvId = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> listState = backend
.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -290,7 +278,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
kvId = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ new TestType.IncompatibleTestTypeSerializer());
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getPartitionedState(VoidNamespace.INSTANCE, CustomVoidNamespaceSerializer.INSTANCE, kvId);
@@ -312,7 +300,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
InternalPriorityQueue<TestType> internalPriorityQueue = backend.create(
- "testPriorityQueue", new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ "testPriorityQueue", new TestType.V1TestTypeSerializer());
internalPriorityQueue.add(new TestType("key-1", 123));
internalPriorityQueue.add(new TestType("key-2", 346));
@@ -325,7 +313,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot);
backend.create(
- "testPriorityQueue", new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ "testPriorityQueue", new TestType.IncompatibleTestTypeSerializer());
Assert.fail("should have failed");
} catch (Exception e) {
@@ -341,7 +329,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
AbstractKeyedStateBackend<TestType> backend = createKeyedBackend(
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
final String stateName = "test-name";
try {
@@ -361,7 +349,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
// the new key serializer is incompatible; this should fail the restore
- restoreKeyedBackend(new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE), snapshot);
+ restoreKeyedBackend(new TestType.IncompatibleTestTypeSerializer(), snapshot);
Assert.fail("should have failed");
} catch (Exception e) {
@@ -370,7 +358,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
// the new key serializer requires migration; this should fail the restore
- restoreKeyedBackend(new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION), snapshot);
+ restoreKeyedBackend(new TestType.V2TestTypeSerializer(), snapshot);
Assert.fail("should have failed");
} catch (Exception e) {
@@ -394,7 +382,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
ValueState<Integer> valueState = backend
.getPartitionedState(
new TestType("namespace", 123),
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS),
+ new TestType.V1TestTypeSerializer(),
kvId);
backend.setCurrentKey(1);
@@ -413,7 +401,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// the new namespace serializer is incompatible; this should fail the restore
backend.getPartitionedState(
new TestType("namespace", 123),
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE),
+ new TestType.IncompatibleTestTypeSerializer(),
kvId);
Assert.fail("should have failed");
@@ -428,7 +416,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// the new namespace serializer requires migration; this should fail the restore
backend.getPartitionedState(
new TestType("namespace", 123),
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION),
+ new TestType.V2TestTypeSerializer(),
kvId);
Assert.fail("should have failed");
@@ -454,7 +442,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> state = backend.getListState(descriptor);
state.add(new TestType("foo", 13));
@@ -468,7 +456,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestType.V2TestTypeSerializer());
state = backend.getListState(descriptor);
// the state backend should have decided whether or not it needs to perform state migration;
@@ -493,7 +481,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> state = backend.getUnionListState(descriptor);
state.add(new TestType("foo", 13));
@@ -507,7 +495,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestType.V2TestTypeSerializer());
state = backend.getUnionListState(descriptor);
// the state backend should have decided whether or not it needs to perform state migration;
@@ -533,7 +521,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
MapStateDescriptor<Integer, TestType> descriptor = new MapStateDescriptor<>(
stateName,
IntSerializer.INSTANCE,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
BroadcastState<Integer, TestType> state = backend.getBroadcastState(descriptor);
state.put(3, new TestType("foo", 13));
@@ -548,7 +536,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new MapStateDescriptor<>(
stateName,
IntSerializer.INSTANCE,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION));
+ new TestType.V2TestTypeSerializer());
state = backend.getBroadcastState(descriptor);
// the state backend should have decided whether or not it needs to perform state migration;
@@ -571,7 +559,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
MapStateDescriptor<TestType, Integer> descriptor = new MapStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS),
+ new TestType.V1TestTypeSerializer(),
IntSerializer.INSTANCE);
BroadcastState<TestType, Integer> state = backend.getBroadcastState(descriptor);
@@ -586,7 +574,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new MapStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.REQUIRES_MIGRATION),
+ new TestType.V2TestTypeSerializer(),
IntSerializer.INSTANCE);
state = backend.getBroadcastState(descriptor);
@@ -610,7 +598,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> state = backend.getListState(descriptor);
state.add(new TestType("foo", 13));
@@ -624,7 +612,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ new TestType.IncompatibleTestTypeSerializer());
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getListState(descriptor);
@@ -647,7 +635,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
ListStateDescriptor<TestType> descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
ListState<TestType> state = backend.getUnionListState(descriptor);
state.add(new TestType("foo", 13));
@@ -661,7 +649,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new ListStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ new TestType.IncompatibleTestTypeSerializer());
// the new serializer is INCOMPATIBLE, so registering the state should fail
backend.getUnionListState(descriptor);
@@ -685,7 +673,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
MapStateDescriptor<Integer, TestType> descriptor = new MapStateDescriptor<>(
stateName,
IntSerializer.INSTANCE,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS));
+ new TestType.V1TestTypeSerializer());
BroadcastState<Integer, TestType> state = backend.getBroadcastState(descriptor);
state.put(3, new TestType("foo", 13));
@@ -700,7 +688,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new MapStateDescriptor<>(
stateName,
IntSerializer.INSTANCE,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE));
+ new TestType.IncompatibleTestTypeSerializer());
// the new value serializer is INCOMPATIBLE, so registering the state should fail
backend.getBroadcastState(descriptor);
@@ -723,7 +711,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
try {
MapStateDescriptor<TestType, Integer> descriptor = new MapStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS),
+ new TestType.V1TestTypeSerializer(),
IntSerializer.INSTANCE);
BroadcastState<TestType, Integer> state = backend.getBroadcastState(descriptor);
@@ -738,7 +726,7 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
descriptor = new MapStateDescriptor<>(
stateName,
- new TestSerializer(SerializerCompatibilityType.INCOMPATIBLE),
+ new TestType.IncompatibleTestTypeSerializer(),
IntSerializer.INSTANCE);
// the new key serializer is INCOMPATIBLE, so registering the state should fail
@@ -756,223 +744,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
// Test types, serializers, and serializer snapshots
// -------------------------------------------------------------------------------
- /**
- * The type used as state under tests.
- *
- * <p>This is implemented so that the type can also be used as keyed priority queue state.
- */
- private static class TestType implements HeapPriorityQueueElement, PriorityComparable<TestType>, Keyed<String> {
-
- private int index;
-
- private final int value;
- private final String key;
-
- public TestType(String key, int value) {
- this.key = key;
- this.value = value;
- }
-
- @Override
- public String getKey() {
- return key;
- }
-
- @Override
- public int comparePriorityTo(@Nonnull TestType other) {
- return Integer.compare(value, other.value);
- }
-
- @Override
- public int getInternalIndex() {
- return index;
- }
-
- @Override
- public void setInternalIndex(int newIndex) {
- this.index = newIndex;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj == null || !(obj instanceof StateBackendMigrationTestBase.TestType)) {
- return false;
- }
-
- if (obj == this) {
- return true;
- }
-
- TestType other = (TestType) obj;
- return Objects.equals(key, other.key) && value == other.value;
- }
-
- @Override
- public int hashCode() {
- return 31 * key.hashCode() + value;
- }
- }
-
- private static class TestSerializer extends TypeSerializer<TestType> {
-
- private static final String MIGRATION_PAYLOAD = "random-migration-payload";
-
- private final SerializerCompatibilityType compatibilityType;
-
- TestSerializer(SerializerCompatibilityType compatibilityType) {
- this.compatibilityType = compatibilityType;
- }
-
- // --------------------------------------------------------------------------------
- // State serialization relevant methods
- // --------------------------------------------------------------------------------
-
- @Override
- public void serialize(TestType record, DataOutputView target) throws IOException {
- switch (compatibilityType) {
- case COMPATIBLE_AS_IS:
- target.writeUTF(record.getKey());
- target.writeInt(record.value);
- break;
-
- case REQUIRES_MIGRATION:
- target.writeUTF(record.getKey());
- target.writeUTF(MIGRATION_PAYLOAD);
- target.writeInt(record.value);
- target.writeBoolean(true);
- break;
-
- case INCOMPATIBLE:
- // the serializer shouldn't be used in this case
- throw new UnsupportedOperationException();
- }
- }
-
- @Override
- public TestType deserialize(DataInputView source) throws IOException {
- String key;
- int value;
-
- switch (compatibilityType) {
- case COMPATIBLE_AS_IS:
- key = source.readUTF();
- value = source.readInt();
- break;
-
- case REQUIRES_MIGRATION:
- key = source.readUTF();
- Assert.assertEquals(MIGRATION_PAYLOAD, source.readUTF());
- value = source.readInt();
- Assert.assertTrue(source.readBoolean());
- break;
-
- default:
- case INCOMPATIBLE:
- // the serializer shouldn't be used in this case
- throw new UnsupportedOperationException();
- }
-
- return new TestType(key, value);
- }
-
- @Override
- public TestType copy(TestType from) {
- return new TestType(from.key, from.value);
- }
-
- @Override
- public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
- return new TestSerializerSnapshot();
- }
-
- // --------------------------------------------------------------------------------
- // Miscellaneous serializer methods
- // --------------------------------------------------------------------------------
-
- @Override
- public void copy(DataInputView source, DataOutputView target) throws IOException {
- serialize(deserialize(source), target);
- }
-
- @Override
- public TestType deserialize(TestType reuse, DataInputView source) throws IOException {
- return deserialize(source);
- }
-
- @Override
- public TestType copy(TestType from, TestType reuse) {
- return copy(from);
- }
-
- @Override
- public TestType createInstance() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public TypeSerializer<TestType> duplicate() {
- return this;
- }
-
- @Override
- public boolean isImmutableType() {
- return false;
- }
-
- @Override
- public int getLength() {
- return -1;
- }
-
- @Override
- public boolean canEqual(Object obj) {
- return getClass().equals(obj.getClass());
- }
-
- @Override
- public int hashCode() {
- return getClass().hashCode();
- }
-
- @Override
- public boolean equals(Object obj) {
- return obj == this;
- }
- }
-
- public static class TestSerializerSnapshot implements TypeSerializerSnapshot<TestType> {
-
- @Override
- public int getCurrentVersion() {
- return 1;
- }
-
- @Override
- public TypeSerializer<TestType> restoreSerializer() {
- return new TestSerializer(SerializerCompatibilityType.COMPATIBLE_AS_IS);
- }
-
- @Override
- public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(TypeSerializer<TestType> newSerializer) {
- switch (((TestSerializer) newSerializer).compatibilityType) {
- case COMPATIBLE_AS_IS:
- return TypeSerializerSchemaCompatibility.compatibleAsIs();
- case REQUIRES_MIGRATION:
- return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- case INCOMPATIBLE:
- return TypeSerializerSchemaCompatibility.incompatible();
- default:
- throw new UnsupportedOperationException();
- }
- }
-
- @Override
- public void writeSnapshot(DataOutputView out) throws IOException {}
-
- @Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {}
- }
-
public static class CustomVoidNamespaceSerializer extends TypeSerializer<VoidNamespace> {
private static final long serialVersionUID = 1L;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
new file mode 100644
index 0000000..e3b0a06
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java
@@ -0,0 +1,237 @@
+/*
+ * 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.testutils.statemigration;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.runtime.state.Keyed;
+import org.apache.flink.runtime.state.PriorityComparable;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement;
+import org.junit.Assert;
+
+import javax.annotation.Nonnull;
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * A data type used as state in state migration tests.
+ *
+ * <p>This is implemented so that the type can also be used as keyed priority queue state.
+ */
+public class TestType implements HeapPriorityQueueElement, PriorityComparable<TestType>, Keyed<String> {
+
+ private int index;
+
+ private final int value;
+ private final String key;
+
+ public TestType(String key, int value) {
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public String getKey() {
+ return key;
+ }
+
+ public int getValue() {
+ return value;
+ }
+
+ @Override
+ public int comparePriorityTo(@Nonnull TestType other) {
+ return Integer.compare(value, other.value);
+ }
+
+ @Override
+ public int getInternalIndex() {
+ return index;
+ }
+
+ @Override
+ public void setInternalIndex(int newIndex) {
+ this.index = newIndex;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null || !(obj instanceof TestType)) {
+ return false;
+ }
+
+ if (obj == this) {
+ return true;
+ }
+
+ TestType other = (TestType) obj;
+ return Objects.equals(key, other.key) && value == other.value;
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * key.hashCode() + value;
+ }
+
+ /**
+ * A serializer that read / writes {@link TestType} in schema version 1.
+ */
+ public static class V1TestTypeSerializer extends TestTypeSerializerBase {
+ private static final long serialVersionUID = 5053346160938769779L;
+
+ @Override
+ public void serialize(TestType record, DataOutputView target) throws IOException {
+ target.writeUTF(record.getKey());
+ target.writeInt(record.getValue());
+ }
+
+ @Override
+ public TestType deserialize(DataInputView source) throws IOException {
+ return new TestType(source.readUTF(), source.readInt());
+ }
+
+ @Override
+ public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
+ return new V1TestTypeSerializerSnapshot();
+ }
+ }
+
+ /**
+ * A serializer that read / writes {@link TestType} in schema version 2.
+ * Migration is required if the state was previously written with {@link V1TestTypeSerializer}.
+ */
+ public static class V2TestTypeSerializer extends TestTypeSerializerBase {
+
+ private static final long serialVersionUID = 7199590310936186578L;
+
+ private static final String RANDOM_PAYLOAD = "random-payload";
+
+ @Override
+ public void serialize(TestType record, DataOutputView target) throws IOException {
+ target.writeUTF(record.getKey());
+ target.writeUTF(RANDOM_PAYLOAD);
+ target.writeInt(record.getValue());
+ target.writeBoolean(true);
+ }
+
+ @Override
+ public TestType deserialize(DataInputView source) throws IOException {
+ String key = source.readUTF();
+ Assert.assertEquals(RANDOM_PAYLOAD, source.readUTF());
+ int value = source.readInt();
+ Assert.assertTrue(source.readBoolean());
+
+ return new TestType(key, value);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
+ return new V1TestTypeSerializerSnapshot();
+ }
+ }
+
+ /**
+ * A serializer that is meant to be incompatible with any of the serializers.
+ */
+ public static class IncompatibleTestTypeSerializer extends TestTypeSerializerBase {
+
+ private static final long serialVersionUID = -2959080770523247215L;
+
+ @Override
+ public void serialize(TestType record, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException("This is an incompatible serializer; shouldn't be used.");
+ }
+
+ @Override
+ public TestType deserialize(DataInputView source) throws IOException {
+ throw new UnsupportedOperationException("This is an incompatible serializer; shouldn't be used.");
+ }
+
+ @Override
+ public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
+ throw new UnsupportedOperationException("This is an incompatible serializer; shouldn't be used.");
+ }
+ }
+
+ public static abstract class TestTypeSerializerBase extends TypeSerializer<TestType> {
+
+ private static final long serialVersionUID = 256299937766275871L;
+
+ // --------------------------------------------------------------------------------
+ // Miscellaneous serializer methods
+ // --------------------------------------------------------------------------------
+
+ @Override
+ public TestType copy(TestType from) {
+ return new TestType(from.getKey(), from.getValue());
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ serialize(deserialize(source), target);
+ }
+
+ @Override
+ public TestType deserialize(TestType reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public TestType copy(TestType from, TestType reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public TestType createInstance() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TypeSerializer<TestType> duplicate() {
+ return this;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return getClass().equals(obj.getClass());
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj == this;
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java
new file mode 100644
index 0000000..b2b802a
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java
@@ -0,0 +1,62 @@
+/*
+ * 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.testutils.statemigration;
+
+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;
+
+/**
+ * Snapshot class for {@link TestType.V1TestTypeSerializer}.
+ */
+public class V1TestTypeSerializerSnapshot implements TypeSerializerSnapshot<TestType> {
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(TypeSerializer<TestType> newSerializer) {
+ if (newSerializer instanceof TestType.V1TestTypeSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ } else if (newSerializer instanceof TestType.V2TestTypeSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ } else {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ @Override
+ public TypeSerializer<TestType> restoreSerializer() {
+ return new TestType.V1TestTypeSerializer();
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java
new file mode 100644
index 0000000..3cd4fff
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java
@@ -0,0 +1,60 @@
+/*
+ * 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.testutils.statemigration;
+
+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;
+
+/**
+ * Snapshot class for {@link TestType.V2TestTypeSerializer}.
+ */
+public class V2TestTypeSerializerSnapshot implements TypeSerializerSnapshot<TestType> {
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(TypeSerializer<TestType> newSerializer) {
+ if (newSerializer instanceof TestType.V2TestTypeSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ } else {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+
+ @Override
+ public TypeSerializer<TestType> restoreSerializer() {
+ return new TestType.V2TestTypeSerializer();
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ }
+}
[flink] 05/15: [FLINK-11094] [state backends] Let meta infos always
lazily access restore serializer
Posted by tz...@apache.org.
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 ebc46c5ec1977ecf70628505eab23c942862857b
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sun Dec 9 15:15:04 2018 +0800
[FLINK-11094] [state backends] Let meta infos always lazily access restore serializer
This commit introduces StateSerializerProvider that
wraps logic on how to obtain serializers for registered state,
either with the previous schema of state in checkpoints or
the current schema of state.
All state meta info subclasses use
StateSerializerProviders to replace direct serializer instances. This
allows meta infos that were instantiated with restored serializer
snapshots to not eagerly access the restore serializer when restoring
state. This needs to be avoided since when restoring from 1.6, the
restore serializer might not be available; for RocksDB, this should be
tolerable.
---
.../RegisteredBroadcastStateBackendMetaInfo.java | 84 +++++--
.../RegisteredKeyValueStateBackendMetaInfo.java | 115 ++++++----
.../RegisteredOperatorStateBackendMetaInfo.java | 46 +++-
...egisteredPriorityQueueStateBackendMetaInfo.java | 39 +++-
.../runtime/state/StateSerializerProvider.java | 245 +++++++++++++++++++++
.../flink/runtime/state/StateBackendTestBase.java | 51 +++++
.../runtime/state/StateSerializerProviderTest.java | 187 ++++++++++++++++
.../streaming/state/RocksDBKeyedStateBackend.java | 125 +++++------
8 files changed, 739 insertions(+), 153 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
index 70a1414..95a650e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredBroadcastStateBackendMetaInfo.java
@@ -19,11 +19,13 @@
package org.apache.flink.runtime.state;
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.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
@@ -38,11 +40,11 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
/** The type serializer for the keys in the map state. */
@Nonnull
- private final TypeSerializer<K> keySerializer;
+ private final StateSerializerProvider<K> keySerializerProvider;
/** The type serializer for the values in the map state. */
@Nonnull
- private final TypeSerializer<V> valueSerializer;
+ private final StateSerializerProvider<V> valueSerializerProvider;
public RegisteredBroadcastStateBackendMetaInfo(
@Nonnull final String name,
@@ -50,19 +52,19 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
@Nonnull final TypeSerializer<K> keySerializer,
@Nonnull final TypeSerializer<V> valueSerializer) {
- super(name);
- Preconditions.checkArgument(assignmentMode == OperatorStateHandle.Mode.BROADCAST);
- this.assignmentMode = assignmentMode;
- this.keySerializer = keySerializer;
- this.valueSerializer = valueSerializer;
+ this(
+ name,
+ assignmentMode,
+ StateSerializerProvider.fromNewState(keySerializer),
+ StateSerializerProvider.fromNewState(valueSerializer));
}
public RegisteredBroadcastStateBackendMetaInfo(@Nonnull RegisteredBroadcastStateBackendMetaInfo<K, V> copy) {
this(
Preconditions.checkNotNull(copy).name,
copy.assignmentMode,
- copy.keySerializer.duplicate(),
- copy.valueSerializer.duplicate());
+ copy.getKeySerializer().duplicate(),
+ copy.getValueSerializer().duplicate());
}
@SuppressWarnings("unchecked")
@@ -71,10 +73,13 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
snapshot.getName(),
OperatorStateHandle.Mode.valueOf(
snapshot.getOption(StateMetaInfoSnapshot.CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE)),
- (TypeSerializer<K>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER)),
- (TypeSerializer<V>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)));
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<K>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER))),
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<V>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
+
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.BROADCAST == snapshot.getBackendStateType());
}
@@ -86,6 +91,19 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
return new RegisteredBroadcastStateBackendMetaInfo<>(this);
}
+ private RegisteredBroadcastStateBackendMetaInfo(
+ @Nonnull final String name,
+ @Nonnull final OperatorStateHandle.Mode assignmentMode,
+ @Nonnull final StateSerializerProvider<K> keySerializerProvider,
+ @Nonnull final StateSerializerProvider<V> valueSerializerProvider) {
+
+ super(name);
+ Preconditions.checkArgument(assignmentMode == OperatorStateHandle.Mode.BROADCAST);
+ this.assignmentMode = assignmentMode;
+ this.keySerializerProvider = keySerializerProvider;
+ this.valueSerializerProvider = valueSerializerProvider;
+ }
+
@Nonnull
@Override
public StateMetaInfoSnapshot snapshot() {
@@ -94,12 +112,32 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
@Nonnull
public TypeSerializer<K> getKeySerializer() {
- return keySerializer;
+ return keySerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<K> updateKeySerializer(TypeSerializer<K> newKeySerializer) {
+ return keySerializerProvider.registerNewSerializerForRestoredState(newKeySerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<K> getPreviousKeySerializer() {
+ return keySerializerProvider.previousSchemaSerializer();
}
@Nonnull
public TypeSerializer<V> getValueSerializer() {
- return valueSerializer;
+ return valueSerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<V> updateValueSerializer(TypeSerializer<V> newValueSerializer) {
+ return valueSerializerProvider.registerNewSerializerForRestoredState(newValueSerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<V> getPreviousValueSerializer() {
+ return valueSerializerProvider.previousSchemaSerializer();
}
@Nonnull
@@ -122,16 +160,16 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
return Objects.equals(name, other.getName())
&& Objects.equals(assignmentMode, other.getAssignmentMode())
- && Objects.equals(keySerializer, other.getKeySerializer())
- && Objects.equals(valueSerializer, other.getValueSerializer());
+ && Objects.equals(getKeySerializer(), other.getKeySerializer())
+ && Objects.equals(getValueSerializer(), other.getValueSerializer());
}
@Override
public int hashCode() {
int result = name.hashCode();
result = 31 * result + assignmentMode.hashCode();
- result = 31 * result + keySerializer.hashCode();
- result = 31 * result + valueSerializer.hashCode();
+ result = 31 * result + getKeySerializer().hashCode();
+ result = 31 * result + getValueSerializer().hashCode();
return result;
}
@@ -139,8 +177,8 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
public String toString() {
return "RegisteredBroadcastBackendStateMetaInfo{" +
"name='" + name + '\'' +
- ", keySerializer=" + keySerializer +
- ", valueSerializer=" + valueSerializer +
+ ", keySerializer=" + getKeySerializer() +
+ ", valueSerializer=" + getValueSerializer() +
", assignmentMode=" + assignmentMode +
'}';
}
@@ -154,8 +192,12 @@ public class RegisteredBroadcastStateBackendMetaInfo<K, V> extends RegisteredSta
Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshotsMap = new HashMap<>(2);
String keySerializerKey = StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER.toString();
String valueSerializerKey = StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString();
+
+ TypeSerializer<K> keySerializer = getKeySerializer();
serializerMap.put(keySerializerKey, keySerializer.duplicate());
serializerConfigSnapshotsMap.put(keySerializerKey, keySerializer.snapshotConfiguration());
+
+ TypeSerializer<V> valueSerializer = getValueSerializer();
serializerMap.put(valueSerializerKey, valueSerializer.duplicate());
serializerConfigSnapshotsMap.put(valueSerializerKey, valueSerializer.snapshotConfiguration());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
index d05f31a..ebe8e94 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.api.common.state.StateDescriptor;
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.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.util.Preconditions;
@@ -44,18 +45,24 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
@Nonnull
private final StateDescriptor.Type stateType;
@Nonnull
- private final TypeSerializer<N> namespaceSerializer;
+ private final StateSerializerProvider<N> namespaceSerializerProvider;
@Nonnull
- private final TypeSerializer<S> stateSerializer;
+ private final StateSerializerProvider<S> stateSerializerProvider;
@Nullable
- private final StateSnapshotTransformer<S> snapshotTransformer;
+ private StateSnapshotTransformer<S> snapshotTransformer;
public RegisteredKeyValueStateBackendMetaInfo(
@Nonnull StateDescriptor.Type stateType,
@Nonnull String name,
@Nonnull TypeSerializer<N> namespaceSerializer,
@Nonnull TypeSerializer<S> stateSerializer) {
- this(stateType, name, namespaceSerializer, stateSerializer, null);
+
+ this(
+ stateType,
+ name,
+ StateSerializerProvider.fromNewState(namespaceSerializer),
+ StateSerializerProvider.fromNewState(stateSerializer),
+ null);
}
public RegisteredKeyValueStateBackendMetaInfo(
@@ -65,11 +72,12 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
@Nonnull TypeSerializer<S> stateSerializer,
@Nullable StateSnapshotTransformer<S> snapshotTransformer) {
- super(name);
- this.stateType = stateType;
- this.namespaceSerializer = namespaceSerializer;
- this.stateSerializer = stateSerializer;
- this.snapshotTransformer = snapshotTransformer;
+ this(
+ stateType,
+ name,
+ StateSerializerProvider.fromNewState(namespaceSerializer),
+ StateSerializerProvider.fromNewState(stateSerializer),
+ snapshotTransformer);
}
@SuppressWarnings("unchecked")
@@ -77,13 +85,31 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
this(
StateDescriptor.Type.valueOf(snapshot.getOption(StateMetaInfoSnapshot.CommonOptionsKeys.KEYED_STATE_TYPE)),
snapshot.getName(),
- (TypeSerializer<N>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER)),
- (TypeSerializer<S>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)), null);
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<N>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER))),
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<S>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
+ null);
+
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.KEY_VALUE == snapshot.getBackendStateType());
}
+ private RegisteredKeyValueStateBackendMetaInfo(
+ @Nonnull StateDescriptor.Type stateType,
+ @Nonnull String name,
+ @Nonnull StateSerializerProvider<N> namespaceSerializerProvider,
+ @Nonnull StateSerializerProvider<S> stateSerializerProvider,
+ @Nullable StateSnapshotTransformer<S> snapshotTransformer) {
+
+ super(name);
+ this.stateType = stateType;
+ this.namespaceSerializerProvider = namespaceSerializerProvider;
+ this.stateSerializerProvider = stateSerializerProvider;
+ this.snapshotTransformer = snapshotTransformer;
+ }
+
@Nonnull
public StateDescriptor.Type getStateType() {
return stateType;
@@ -91,12 +117,32 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
@Nonnull
public TypeSerializer<N> getNamespaceSerializer() {
- return namespaceSerializer;
+ return namespaceSerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<N> updateNamespaceSerializer(TypeSerializer<N> newNamespaceSerializer) {
+ return namespaceSerializerProvider.registerNewSerializerForRestoredState(newNamespaceSerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<N> getPreviousNamespaceSerializer() {
+ return namespaceSerializerProvider.previousSchemaSerializer();
}
@Nonnull
public TypeSerializer<S> getStateSerializer() {
- return stateSerializer;
+ return stateSerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<S> updateStateSerializer(TypeSerializer<S> newStateSerializer) {
+ return stateSerializerProvider.registerNewSerializerForRestoredState(newStateSerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<S> getPreviousStateSerializer() {
+ return stateSerializerProvider.previousSchemaSerializer();
}
@Nullable
@@ -104,6 +150,10 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
return snapshotTransformer;
}
+ public void updateSnapshotTransformer(StateSnapshotTransformer<S> snapshotTransformer) {
+ this.snapshotTransformer = snapshotTransformer;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -133,8 +183,8 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
return "RegisteredKeyedBackendStateMetaInfo{" +
"stateType=" + stateType +
", name='" + name + '\'' +
- ", namespaceSerializer=" + namespaceSerializer +
- ", stateSerializer=" + stateSerializer +
+ ", namespaceSerializer=" + getNamespaceSerializer() +
+ ", stateSerializer=" + getStateSerializer() +
'}';
}
@@ -153,34 +203,19 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
return computeSnapshot();
}
- public static void checkStateMetaInfo(StateMetaInfoSnapshot stateMetaInfoSnapshot, StateDescriptor<?, ?> stateDesc) {
- Preconditions.checkState(
- stateMetaInfoSnapshot != null,
- "Requested to check compatibility of a restored RegisteredKeyedBackendStateMetaInfo," +
- " but its corresponding restored snapshot cannot be found.");
-
- Preconditions.checkState(stateMetaInfoSnapshot.getBackendStateType()
- == StateMetaInfoSnapshot.BackendStateType.KEY_VALUE,
- "Incompatible state types. " +
- "Was [" + stateMetaInfoSnapshot.getBackendStateType() + "], " +
- "registered as [" + StateMetaInfoSnapshot.BackendStateType.KEY_VALUE + "].");
+ public void checkStateMetaInfo(StateDescriptor<?, ?> stateDesc) {
Preconditions.checkState(
- Objects.equals(stateDesc.getName(), stateMetaInfoSnapshot.getName()),
+ Objects.equals(stateDesc.getName(), getName()),
"Incompatible state names. " +
- "Was [" + stateMetaInfoSnapshot.getName() + "], " +
+ "Was [" + getName() + "], " +
"registered with [" + stateDesc.getName() + "].");
- final StateDescriptor.Type restoredType =
- StateDescriptor.Type.valueOf(
- stateMetaInfoSnapshot.getOption(
- StateMetaInfoSnapshot.CommonOptionsKeys.KEYED_STATE_TYPE));
-
- if (stateDesc.getType() != StateDescriptor.Type.UNKNOWN && restoredType != StateDescriptor.Type.UNKNOWN) {
+ if (stateDesc.getType() != StateDescriptor.Type.UNKNOWN && getStateType() != StateDescriptor.Type.UNKNOWN) {
Preconditions.checkState(
- stateDesc.getType() == restoredType,
+ stateDesc.getType() == getStateType(),
"Incompatible key/value state types. " +
- "Was [" + restoredType + "], " +
+ "Was [" + getStateType() + "], " +
"registered with [" + stateDesc.getType() + "].");
}
}
@@ -194,8 +229,12 @@ public class RegisteredKeyValueStateBackendMetaInfo<N, S> extends RegisteredStat
Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshotsMap = new HashMap<>(2);
String namespaceSerializerKey = StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString();
String valueSerializerKey = StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString();
+
+ TypeSerializer<N> namespaceSerializer = getNamespaceSerializer();
serializerMap.put(namespaceSerializerKey, namespaceSerializer.duplicate());
serializerConfigSnapshotsMap.put(namespaceSerializerKey, namespaceSerializer.snapshotConfiguration());
+
+ TypeSerializer<S> stateSerializer = getStateSerializer();
serializerMap.put(valueSerializerKey, stateSerializer.duplicate());
serializerConfigSnapshotsMap.put(valueSerializerKey, stateSerializer.snapshotConfiguration());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
index 10ba029..afb3d77 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorStateBackendMetaInfo.java
@@ -19,11 +19,13 @@
package org.apache.flink.runtime.state;
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.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.Map;
@@ -46,21 +48,22 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
* The type serializer for the elements in the state list
*/
@Nonnull
- private final TypeSerializer<S> partitionStateSerializer;
+ private final StateSerializerProvider<S> partitionStateSerializerProvider;
public RegisteredOperatorStateBackendMetaInfo(
@Nonnull String name,
@Nonnull TypeSerializer<S> partitionStateSerializer,
@Nonnull OperatorStateHandle.Mode assignmentMode) {
- super(name);
- this.partitionStateSerializer = partitionStateSerializer;
- this.assignmentMode = assignmentMode;
+ this(
+ name,
+ StateSerializerProvider.fromNewState(partitionStateSerializer),
+ assignmentMode);
}
private RegisteredOperatorStateBackendMetaInfo(@Nonnull RegisteredOperatorStateBackendMetaInfo<S> copy) {
this(
Preconditions.checkNotNull(copy).name,
- copy.partitionStateSerializer.duplicate(),
+ copy.getPartitionStateSerializer().duplicate(),
copy.assignmentMode);
}
@@ -68,13 +71,24 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
public RegisteredOperatorStateBackendMetaInfo(@Nonnull StateMetaInfoSnapshot snapshot) {
this(
snapshot.getName(),
- (TypeSerializer<S>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)),
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<S>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))),
OperatorStateHandle.Mode.valueOf(
snapshot.getOption(StateMetaInfoSnapshot.CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE)));
+
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.OPERATOR == snapshot.getBackendStateType());
}
+ private RegisteredOperatorStateBackendMetaInfo(
+ @Nonnull String name,
+ @Nonnull StateSerializerProvider<S> partitionStateSerializerProvider,
+ @Nonnull OperatorStateHandle.Mode assignmentMode) {
+ super(name);
+ this.partitionStateSerializerProvider = partitionStateSerializerProvider;
+ this.assignmentMode = assignmentMode;
+ }
+
/**
* Creates a deep copy of the itself.
*/
@@ -96,7 +110,17 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
@Nonnull
public TypeSerializer<S> getPartitionStateSerializer() {
- return partitionStateSerializer;
+ return partitionStateSerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<S> updatePartitionStateSerializer(TypeSerializer<S> newPartitionStateSerializer) {
+ return partitionStateSerializerProvider.registerNewSerializerForRestoredState(newPartitionStateSerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<S> getPreviousPartitionStateSerializer() {
+ return partitionStateSerializerProvider.previousSchemaSerializer();
}
@Override
@@ -112,7 +136,7 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
return (obj instanceof RegisteredOperatorStateBackendMetaInfo)
&& name.equals(((RegisteredOperatorStateBackendMetaInfo) obj).getName())
&& assignmentMode.equals(((RegisteredOperatorStateBackendMetaInfo) obj).getAssignmentMode())
- && partitionStateSerializer.equals(((RegisteredOperatorStateBackendMetaInfo) obj).getPartitionStateSerializer());
+ && getPartitionStateSerializer().equals(((RegisteredOperatorStateBackendMetaInfo) obj).getPartitionStateSerializer());
}
@Override
@@ -128,7 +152,7 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
return "RegisteredOperatorBackendStateMetaInfo{" +
"name='" + name + "\'" +
", assignmentMode=" + assignmentMode +
- ", partitionStateSerializer=" + partitionStateSerializer +
+ ", partitionStateSerializer=" + getPartitionStateSerializer() +
'}';
}
@@ -138,6 +162,8 @@ public class RegisteredOperatorStateBackendMetaInfo<S> extends RegisteredStateMe
StateMetaInfoSnapshot.CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE.toString(),
assignmentMode.toString());
String valueSerializerKey = StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString();
+
+ TypeSerializer<S> partitionStateSerializer = getPartitionStateSerializer();
Map<String, TypeSerializer<?>> serializerMap =
Collections.singletonMap(valueSerializerKey, partitionStateSerializer.duplicate());
Map<String, TypeSerializerSnapshot<?>> serializerConfigSnapshotsMap =
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
index 0304b92..60c88e3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredPriorityQueueStateBackendMetaInfo.java
@@ -19,11 +19,13 @@
package org.apache.flink.runtime.state;
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.runtime.state.metainfo.StateMetaInfoSnapshot;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.Map;
@@ -34,24 +36,34 @@ import java.util.Map;
public class RegisteredPriorityQueueStateBackendMetaInfo<T> extends RegisteredStateMetaInfoBase {
@Nonnull
- private final TypeSerializer<T> elementSerializer;
+ private final StateSerializerProvider<T> elementSerializerProvider;
public RegisteredPriorityQueueStateBackendMetaInfo(
@Nonnull String name,
@Nonnull TypeSerializer<T> elementSerializer) {
- super(name);
- this.elementSerializer = elementSerializer;
+ this(name, StateSerializerProvider.fromNewState(elementSerializer));
}
@SuppressWarnings("unchecked")
public RegisteredPriorityQueueStateBackendMetaInfo(StateMetaInfoSnapshot snapshot) {
- this(snapshot.getName(),
- (TypeSerializer<T>) Preconditions.checkNotNull(
- snapshot.restoreTypeSerializer(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)));
+ this(
+ snapshot.getName(),
+ StateSerializerProvider.fromRestoredState(
+ (TypeSerializerSnapshot<T>) Preconditions.checkNotNull(
+ snapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER))));
+
Preconditions.checkState(StateMetaInfoSnapshot.BackendStateType.PRIORITY_QUEUE == snapshot.getBackendStateType());
}
+ private RegisteredPriorityQueueStateBackendMetaInfo(
+ @Nonnull String name,
+ @Nonnull StateSerializerProvider<T> elementSerializerProvider) {
+
+ super(name);
+ this.elementSerializerProvider = elementSerializerProvider;
+ }
+
@Nonnull
@Override
public StateMetaInfoSnapshot snapshot() {
@@ -60,10 +72,21 @@ public class RegisteredPriorityQueueStateBackendMetaInfo<T> extends RegisteredSt
@Nonnull
public TypeSerializer<T> getElementSerializer() {
- return elementSerializer;
+ return elementSerializerProvider.currentSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<T> updateElementSerializer(TypeSerializer<T> newElementSerializer) {
+ return elementSerializerProvider.registerNewSerializerForRestoredState(newElementSerializer);
+ }
+
+ @Nullable
+ public TypeSerializer<T> getPreviousElementSerializer() {
+ return elementSerializerProvider.previousSchemaSerializer();
}
private StateMetaInfoSnapshot computeSnapshot() {
+ TypeSerializer<T> elementSerializer = getElementSerializer();
Map<String, TypeSerializer<?>> serializerMap =
Collections.singletonMap(
StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(),
@@ -82,6 +105,6 @@ public class RegisteredPriorityQueueStateBackendMetaInfo<T> extends RegisteredSt
}
public RegisteredPriorityQueueStateBackendMetaInfo deepCopy() {
- return new RegisteredPriorityQueueStateBackendMetaInfo<>(name, elementSerializer.duplicate());
+ return new RegisteredPriorityQueueStateBackendMetaInfo<>(name, getElementSerializer().duplicate());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateSerializerProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateSerializerProvider.java
new file mode 100644
index 0000000..a24f12e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/StateSerializerProvider.java
@@ -0,0 +1,245 @@
+/*
+ * 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.annotation.Internal;
+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.api.common.typeutils.UnloadableDummyTypeSerializer;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link StateSerializerProvider} wraps logic on how to obtain serializers for registered state,
+ * either with the previous schema of state in checkpoints or the current schema of state.
+ *
+ * @param <T> the type of the state.
+ */
+@Internal
+public abstract class StateSerializerProvider<T> {
+
+ /**
+ * The registered serializer for the state.
+ *
+ * <p>In the case that this provider was created from a restored serializer snapshot via
+ * {@link #fromRestoredState(TypeSerializerSnapshot)}, but a new serializer was never registered
+ * for the state (i.e., this is the case if a restored state was never accessed), this would be {@code null}.
+ */
+ @Nullable
+ TypeSerializer<T> registeredSerializer;
+
+ /**
+ * Creates a {@link StateSerializerProvider} for restored state from the previous serializer's snapshot.
+ *
+ * <p>Once a new serializer is registered for the state, it should be provided via
+ * the {@link #registerNewSerializerForRestoredState(TypeSerializer)} method.
+ *
+ * @param stateSerializerSnapshot the previous serializer's snapshot.
+ * @param <T> the type of the state.
+ *
+ * @return a new {@link StateSerializerProvider} for restored state.
+ */
+ public static <T> StateSerializerProvider<T> fromRestoredState(TypeSerializerSnapshot<T> stateSerializerSnapshot) {
+ return new RestoredStateSerializerProvider<>(stateSerializerSnapshot);
+ }
+
+ /**
+ * Creates a {@link StateSerializerProvider} for new state from the registered state serializer.
+ *
+ * @param registeredStateSerializer the new state's registered serializer.
+ * @param <T> the type of the state.
+ *
+ * @return a new {@link StateSerializerProvider} for new state.
+ */
+ public static <T> StateSerializerProvider<T> fromNewState(TypeSerializer<T> registeredStateSerializer) {
+ return new NewStateSerializerProvider<>(registeredStateSerializer);
+ }
+
+ private StateSerializerProvider(@Nullable TypeSerializer<T> stateSerializer) {
+ this.registeredSerializer = stateSerializer;
+ }
+
+ /**
+ * Gets the serializer that recognizes the current serialization schema of the state.
+ * This is the serializer that should be used for regular state serialization and
+ * deserialization after state has been restored.
+ *
+ * <p>If this provider was created from a restored state's serializer snapshot, while a
+ * new serializer (with a new schema) was not registered for the state (i.e., because
+ * the state was never accessed after it was restored), then the schema of state remains
+ * identical. Therefore, in this case, it is guaranteed that the serializer returned by
+ * this method is the same as the one returned by {@link #previousSchemaSerializer()}.
+ *
+ * <p>If this provider was created from new state, then this always returns the
+ * serializer that the new state was registered with.
+ *
+ * @return a serializer that reads and writes in the current schema of the state.
+ */
+ @Nonnull
+ public abstract TypeSerializer<T> currentSchemaSerializer();
+
+ /**
+ * Gets the serializer that recognizes the previous serialization schema of the state.
+ * This is the serializer that should be used for restoring the state, i.e. when the state
+ * is still in the previous serialization schema.
+ *
+ * <p>This method can only be used if this provider was created from a restored state's serializer
+ * snapshot. If this provider was created from new state, then this method is
+ * irrelevant, since there doesn't exist any previous version of the state schema.
+ *
+ * @return a serializer that reads and writes in the previous schema of the state.
+ */
+ @Nonnull
+ public abstract TypeSerializer<T> previousSchemaSerializer();
+
+ /**
+ * For restored state, register a new serializer that potentially has a new serialization schema.
+ *
+ * <p>Users are allowed to register serializers for state only once. Therefore, this method
+ * is irrelevant if this provider was created from new state, since a state serializer had
+ * been registered already.
+ *
+ * <p>For the case where this provider was created from restored state, then this method should
+ * be called at most once. The new serializer will be checked for its schema compatibility with the
+ * previous serializer's schema, and returned to the caller. The caller is responsible for
+ * checking the result and react appropriately to it, as follows:
+ * <ul>
+ * <li>{@link TypeSerializerSchemaCompatibility#isCompatibleAsIs()}: nothing needs to be done.
+ * {@link #currentSchemaSerializer()} now returns the newly registered serializer.</li>
+ * <li>{@link TypeSerializerSchemaCompatibility#isCompatibleAfterMigration()} ()}: state needs to be
+ * migrated before the serializer returned by {@link #currentSchemaSerializer()} can be used.
+ * The migration should be performed by reading the state with {@link #previousSchemaSerializer()},
+ * and then writing it again with {@link #currentSchemaSerializer()}.</li>
+ * <li>{@link TypeSerializerSchemaCompatibility#isIncompatible()}: the registered serializer is
+ * incompatible. {@link #currentSchemaSerializer()} can no longer return a serializer for
+ * the state, and therefore this provider shouldn't be used anymore.</li>
+ * </ul>
+ *
+ * @return the schema compatibility of the new registered serializer, with respect to the previous serializer.
+ */
+ @Nonnull
+ public abstract TypeSerializerSchemaCompatibility<T> registerNewSerializerForRestoredState(TypeSerializer<T> newSerializer);
+
+ /**
+ * Implementation of the {@link StateSerializerProvider} for the restored state case.
+ */
+ private static class RestoredStateSerializerProvider<T> extends StateSerializerProvider<T> {
+
+ /**
+ * The snapshot of the previous serializer of the state.
+ */
+ @Nonnull
+ private final TypeSerializerSnapshot<T> previousSerializerSnapshot;
+
+ private boolean isRegisteredWithIncompatibleSerializer = false;
+
+ RestoredStateSerializerProvider(TypeSerializerSnapshot<T> previousSerializerSnapshot) {
+ super(null);
+ this.previousSerializerSnapshot = Preconditions.checkNotNull(previousSerializerSnapshot);
+ }
+
+ /**
+ * The restore serializer, lazily created only when the restore serializer is accessed.
+ *
+ * <p>NOTE: It is important to only create this lazily, so that off-heap
+ * state do not fail eagerly when restoring state that has a
+ * {@link UnloadableDummyTypeSerializer} as the previous serializer. This should
+ * be relevant only for restores from Flink versions prior to 1.7.x.
+ */
+ @Nullable
+ private TypeSerializer<T> cachedRestoredSerializer;
+
+ @Override
+ @Nonnull
+ public TypeSerializer<T> currentSchemaSerializer() {
+ if (registeredSerializer != null) {
+ checkState(
+ !isRegisteredWithIncompatibleSerializer,
+ "Unable to provide a serializer with the current schema, because the restored state was " +
+ "registered with a new serializer that has incompatible schema.");
+
+ return registeredSerializer;
+ }
+
+ // if we are not yet registered with a new serializer,
+ // we can just use the restore serializer to read / write the state.
+ return previousSchemaSerializer();
+ }
+
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<T> registerNewSerializerForRestoredState(TypeSerializer<T> newSerializer) {
+ checkNotNull(newSerializer);
+ if (registeredSerializer != null) {
+ throw new UnsupportedOperationException("A serializer has already been registered for the state; re-registration is not allowed.");
+ }
+
+ TypeSerializerSchemaCompatibility<T> result = previousSerializerSnapshot.resolveSchemaCompatibility(newSerializer);
+ if (result.isIncompatible()) {
+ this.isRegisteredWithIncompatibleSerializer = true;
+ }
+ this.registeredSerializer = newSerializer;
+ return result;
+ }
+
+ @Nonnull
+ public final TypeSerializer<T> previousSchemaSerializer() {
+ if (cachedRestoredSerializer != null) {
+ return cachedRestoredSerializer;
+ }
+
+ this.cachedRestoredSerializer = previousSerializerSnapshot.restoreSerializer();
+ return cachedRestoredSerializer;
+ }
+ }
+
+ /**
+ * Implementation of the {@link StateSerializerProvider} for the new state case.
+ */
+ private static class NewStateSerializerProvider<T> extends StateSerializerProvider<T> {
+
+ NewStateSerializerProvider(TypeSerializer<T> registeredStateSerializer) {
+ super(Preconditions.checkNotNull(registeredStateSerializer));
+ }
+
+ @Override
+ @Nonnull
+ @SuppressWarnings("ConstantConditions")
+ public TypeSerializer<T> currentSchemaSerializer() {
+ return registeredSerializer;
+ }
+
+ @Override
+ @Nonnull
+ public TypeSerializerSchemaCompatibility<T> registerNewSerializerForRestoredState(TypeSerializer<T> newSerializer) {
+ throw new UnsupportedOperationException("A serializer has already been registered for the state; re-registration is not allowed.");
+ }
+
+ @Override
+ @Nonnull
+ public TypeSerializer<T> previousSchemaSerializer() {
+ throw new UnsupportedOperationException("This is a NewStateSerializerProvider; you cannot get a restore serializer because there was no restored state.");
+ }
+ }
+}
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 54cac1e..8708613 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
@@ -3145,6 +3145,57 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
backend.dispose();
}
+ @Test
+ public void testSnapshotNonAccessedState() throws Exception {
+ CheckpointStreamFactory streamFactory = createStreamFactory();
+ SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
+ AbstractKeyedStateBackend<String> backend = createKeyedBackend(StringSerializer.INSTANCE);
+
+ final String stateName = "test-name";
+ try {
+ MapStateDescriptor<Integer, String> kvId = new MapStateDescriptor<>(stateName, Integer.class, String.class);
+ MapState<Integer, String> mapState = backend
+ .getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
+
+ // write some state to be snapshotted
+ backend.setCurrentKey("1");
+ mapState.put(11, "foo");
+ backend.setCurrentKey("2");
+ mapState.put(8, "bar");
+ backend.setCurrentKey("3");
+ mapState.put(91, "hello world");
+
+ // take a snapshot, and then restore backend with snapshot
+ KeyedStateHandle snapshot = runSnapshot(
+ backend.snapshot(1L, 2L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
+ backend.dispose();
+
+ backend = restoreKeyedBackend(StringSerializer.INSTANCE, snapshot);
+
+ // now take a snapshot again without accessing the state
+ snapshot = runSnapshot(
+ backend.snapshot(2L, 3L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
+ backend.dispose();
+
+ // we restore again and try to access previous state
+ backend = restoreKeyedBackend(StringSerializer.INSTANCE, snapshot);
+ mapState = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
+
+ backend.setCurrentKey("1");
+ assertEquals("foo", mapState.get(11));
+ backend.setCurrentKey("2");
+ assertEquals("bar", mapState.get(8));
+ backend.setCurrentKey("3");
+ assertEquals("hello world", mapState.get(91));
+
+ snapshot.discardState();
+ } finally {
+ backend.dispose();
+ }
+ }
+
/**
* This test verifies that state is correctly assigned to key groups and that restore
* restores the relevant key groups in the backend.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
new file mode 100644
index 0000000..de1f2bd
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.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 org.apache.flink.runtime.testutils.statemigration.TestType;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Test suit for {@link StateSerializerProvider}.
+ */
+public class StateSerializerProviderTest {
+
+ // --------------------------------------------------------------------------------
+ // Tests for #currentSchemaSerializer()
+ // --------------------------------------------------------------------------------
+
+ @Test
+ public void testCurrentSchemaSerializerForNewStateSerializerProvider() {
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromNewState(new TestType.V1TestTypeSerializer());
+ assertTrue(testProvider.currentSchemaSerializer() instanceof TestType.V1TestTypeSerializer);
+ }
+
+ @Test
+ public void testCurrentSchemaSerializerForRestoredStateSerializerProvider() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+ assertTrue(testProvider.currentSchemaSerializer() instanceof TestType.V1TestTypeSerializer);
+ }
+
+ // --------------------------------------------------------------------------------
+ // Tests for #previousSchemaSerializer()
+ // --------------------------------------------------------------------------------
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testPreviousSchemaSerializerForNewStateSerializerProvider() {
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromNewState(new TestType.V1TestTypeSerializer());
+
+ // this should fail with an exception
+ testProvider.previousSchemaSerializer();
+ }
+
+ @Test
+ public void testPreviousSchemaSerializerForRestoredStateSerializerProvider() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+ assertTrue(testProvider.previousSchemaSerializer() instanceof TestType.V1TestTypeSerializer);
+ }
+
+ @Test
+ public void testLazyInstantiationOfPreviousSchemaSerializer() {
+ // create the provider with an exception throwing snapshot;
+ // this would throw an exception if the restore serializer was eagerly accessed
+ StateSerializerProvider<String> testProvider =
+ StateSerializerProvider.fromRestoredState(new ExceptionThrowingSerializerSnapshot());
+
+ try {
+ // if we fail here, that means the restore serializer was indeed lazily accessed
+ testProvider.previousSchemaSerializer();
+ fail("expected to fail when accessing the restore serializer.");
+ } catch (Exception expected) {
+ // success
+ }
+ }
+
+ // --------------------------------------------------------------------------------
+ // Tests for #registerNewSerializerForRestoredState(TypeSerializer)
+ // --------------------------------------------------------------------------------
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRegisterNewSerializerWithNewStateSerializerProviderShouldFail() {
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromNewState(new TestType.V1TestTypeSerializer());
+ testProvider.registerNewSerializerForRestoredState(new TestType.V2TestTypeSerializer());
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testRegisterNewSerializerTwiceWithNewStateSerializerProviderShouldFail() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+
+ testProvider.registerNewSerializerForRestoredState(new TestType.V2TestTypeSerializer());
+
+ // second registration should fail
+ testProvider.registerNewSerializerForRestoredState(new TestType.V2TestTypeSerializer());
+ }
+
+ @Test
+ public void testRegisterNewCompatibleAsIsSerializer() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+
+ // register compatible serializer for state
+ TypeSerializerSchemaCompatibility<TestType> schemaCompatibility =
+ testProvider.registerNewSerializerForRestoredState(new TestType.V1TestTypeSerializer());
+ assertTrue(schemaCompatibility.isCompatibleAsIs());
+
+ assertTrue(testProvider.currentSchemaSerializer() instanceof TestType.V1TestTypeSerializer);
+ assertTrue(testProvider.previousSchemaSerializer() instanceof TestType.V1TestTypeSerializer);
+ }
+
+ @Test
+ public void testRegisterNewCompatibleAfterMigrationSerializer() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+
+ // register serializer that requires migration for state
+ TypeSerializerSchemaCompatibility<TestType> schemaCompatibility =
+ testProvider.registerNewSerializerForRestoredState(new TestType.V2TestTypeSerializer());
+ assertTrue(schemaCompatibility.isCompatibleAfterMigration());
+ }
+
+ @Test
+ public void testRegisterIncompatibleSerializer() {
+ TestType.V1TestTypeSerializer serializer = new TestType.V1TestTypeSerializer();
+ StateSerializerProvider<TestType> testProvider = StateSerializerProvider.fromRestoredState(serializer.snapshotConfiguration());
+
+ // register serializer that requires migration for state
+ TypeSerializerSchemaCompatibility<TestType> schemaCompatibility =
+ testProvider.registerNewSerializerForRestoredState(new TestType.IncompatibleTestTypeSerializer());
+ assertTrue(schemaCompatibility.isIncompatible());
+
+ try {
+ // a serializer for the current schema will no longer be accessible
+ testProvider.currentSchemaSerializer();
+ } catch (Exception excepted) {
+ // success
+ }
+ }
+
+ // --------------------------------------------------------------------------------
+ // Utilities
+ // --------------------------------------------------------------------------------
+
+ public static class ExceptionThrowingSerializerSnapshot implements TypeSerializerSnapshot<String> {
+
+ @Override
+ public TypeSerializer<String> restoreSerializer() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<String> resolveSchemaCompatibility(TypeSerializer<String> newSerializer) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
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 d12370b..065213b 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
@@ -29,7 +29,6 @@ import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
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.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
@@ -1377,78 +1376,69 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
TypeSerializer<N> namespaceSerializer,
@Nullable StateSnapshotTransformer<SV> snapshotTransformer) throws Exception {
- Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase> stateInfo =
+ Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase> oldStateInfo =
kvStateInformation.get(stateDesc.getName());
TypeSerializer<SV> stateSerializer = stateDesc.getSerializer();
- RegisteredKeyValueStateBackendMetaInfo<N, SV> newMetaInfo = new RegisteredKeyValueStateBackendMetaInfo<>(
- stateDesc.getType(),
- stateDesc.getName(),
- namespaceSerializer,
- stateSerializer,
- snapshotTransformer);
-
- if (stateInfo != null) {
- newMetaInfo = migrateStateIfNecessary(
- newMetaInfo,
+
+ ColumnFamilyHandle newColumnFamily;
+ RegisteredKeyValueStateBackendMetaInfo<N, SV> newMetaInfo;
+ if (oldStateInfo != null) {
+ @SuppressWarnings("unchecked")
+ RegisteredKeyValueStateBackendMetaInfo<N, SV> castedMetaInfo = (RegisteredKeyValueStateBackendMetaInfo<N, SV>) oldStateInfo.f1;
+
+ newMetaInfo = updateRestoredStateMetaInfo(
+ Tuple2.of(oldStateInfo.f0, castedMetaInfo),
stateDesc,
namespaceSerializer,
stateSerializer,
- stateInfo);
+ snapshotTransformer);
- stateInfo.f1 = newMetaInfo;
+ oldStateInfo.f1 = newMetaInfo;
+ newColumnFamily = oldStateInfo.f0;
} else {
- ColumnFamilyHandle columnFamily = createColumnFamily(stateDesc.getName());
+ newMetaInfo = new RegisteredKeyValueStateBackendMetaInfo<>(
+ stateDesc.getType(),
+ stateDesc.getName(),
+ namespaceSerializer,
+ stateSerializer,
+ snapshotTransformer);
- stateInfo = Tuple2.of(columnFamily, newMetaInfo);
- registerKvStateInformation(stateDesc.getName(), stateInfo);
+ newColumnFamily = createColumnFamily(stateDesc.getName());
+ registerKvStateInformation(stateDesc.getName(), Tuple2.of(newColumnFamily, newMetaInfo));
}
- return Tuple2.of(stateInfo.f0, newMetaInfo);
+ return Tuple2.of(newColumnFamily, newMetaInfo);
}
- private <N, S extends State, SV> RegisteredKeyValueStateBackendMetaInfo<N, SV> migrateStateIfNecessary(
- RegisteredKeyValueStateBackendMetaInfo<N, SV> newMetaInfo,
+ private <N, S extends State, SV> RegisteredKeyValueStateBackendMetaInfo<N, SV> updateRestoredStateMetaInfo(
+ Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>> oldStateInfo,
StateDescriptor<S, SV> stateDesc,
TypeSerializer<N> namespaceSerializer,
TypeSerializer<SV> stateSerializer,
- Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase> stateInfo) throws Exception {
-
- StateMetaInfoSnapshot restoredMetaInfoSnapshot = restoredKvStateMetaInfos.get(stateDesc.getName());
-
- Preconditions.checkState(
- restoredMetaInfoSnapshot != null,
- "Requested to check compatibility of a restored RegisteredKeyedBackendStateMetaInfo," +
- " but its corresponding restored snapshot cannot be found.");
+ @Nullable StateSnapshotTransformer<SV> snapshotTransformer) throws Exception {
@SuppressWarnings("unchecked")
- TypeSerializerSnapshot<N> namespaceSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<N>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
- StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString()));
+ RegisteredKeyValueStateBackendMetaInfo<N, SV> restoredKvStateMetaInfo = oldStateInfo.f1;
+
+ restoredKvStateMetaInfo.updateSnapshotTransformer(snapshotTransformer);
- TypeSerializerSchemaCompatibility<N> namespaceCompatibility =
- namespaceSerializerSnapshot.resolveSchemaCompatibility(namespaceSerializer);
- if (!namespaceCompatibility.isCompatibleAsIs()) {
+ TypeSerializerSchemaCompatibility<N> s = restoredKvStateMetaInfo.updateNamespaceSerializer(namespaceSerializer);
+ if (!s.isCompatibleAsIs()) {
throw new StateMigrationException("The new namespace serializer must be compatible.");
}
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<SV> stateSerializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<SV>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString()));
-
- RegisteredKeyValueStateBackendMetaInfo.checkStateMetaInfo(restoredMetaInfoSnapshot, stateDesc);
+ restoredKvStateMetaInfo.checkStateMetaInfo(stateDesc);
- TypeSerializerSchemaCompatibility<SV> stateCompatibility =
- stateSerializerSnapshot.resolveSchemaCompatibility(stateSerializer);
-
- if (stateCompatibility.isCompatibleAfterMigration()) {
- migrateStateValues(stateDesc, stateInfo, restoredMetaInfoSnapshot, newMetaInfo, stateSerializer);
- } else if (stateCompatibility.isIncompatible()) {
+ TypeSerializerSchemaCompatibility<SV> newStateSerializerCompatibility =
+ restoredKvStateMetaInfo.updateStateSerializer(stateSerializer);
+ if (newStateSerializerCompatibility.isCompatibleAfterMigration()) {
+ migrateStateValues(stateDesc, oldStateInfo);
+ } else if (newStateSerializerCompatibility.isIncompatible()) {
throw new StateMigrationException("The new state serializer cannot be incompatible.");
}
- return newMetaInfo;
+ return restoredKvStateMetaInfo;
}
/**
@@ -1458,10 +1448,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
*/
private <N, S extends State, SV> void migrateStateValues(
StateDescriptor<S, SV> stateDesc,
- Tuple2<ColumnFamilyHandle, RegisteredStateMetaInfoBase> stateInfo,
- StateMetaInfoSnapshot restoredMetaInfoSnapshot,
- RegisteredKeyValueStateBackendMetaInfo<N, SV> newMetaInfo,
- TypeSerializer<SV> newStateSerializer) throws Exception {
+ Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N, SV>> stateMetaInfo) throws Exception {
if (stateDesc.getType() == StateDescriptor.Type.MAP) {
throw new StateMigrationException("The new serializer for a MapState requires state migration in order for the job to proceed." +
@@ -1483,7 +1470,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
}
State state = stateFactory.createState(
stateDesc,
- Tuple2.of(stateInfo.f0, newMetaInfo),
+ stateMetaInfo,
RocksDBKeyedStateBackend.this);
if (!(state instanceof AbstractRocksDBState)) {
throw new FlinkRuntimeException(
@@ -1495,16 +1482,11 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
Snapshot rocksDBSnapshot = db.getSnapshot();
try (
- RocksIteratorWrapper iterator = getRocksIterator(db, stateInfo.f0);
+ RocksIteratorWrapper iterator = getRocksIterator(db, stateMetaInfo.f0);
RocksDBWriteBatchWrapper batchWriter = new RocksDBWriteBatchWrapper(db, getWriteOptions())
) {
iterator.seekToFirst();
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<SV> priorValueSerializerSnapshot = (TypeSerializerSnapshot<SV>)
- Preconditions.checkNotNull(restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER));
- TypeSerializer<SV> priorValueSerializer = priorValueSerializerSnapshot.restoreSerializer();
-
DataInputDeserializer serializedValueInput = new DataInputDeserializer();
DataOutputSerializer migratedSerializedValueOutput = new DataOutputSerializer(512);
while (iterator.isValid()) {
@@ -1513,10 +1495,10 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
rocksDBState.migrateSerializedValue(
serializedValueInput,
migratedSerializedValueOutput,
- priorValueSerializer,
- newStateSerializer);
+ stateMetaInfo.f1.getPreviousStateSerializer(),
+ stateMetaInfo.f1.getStateSerializer());
- batchWriter.put(stateInfo.f0, iterator.key(), migratedSerializedValueOutput.getCopyOfBuffer());
+ batchWriter.put(stateMetaInfo.f0, iterator.key(), migratedSerializedValueOutput.getCopyOfBuffer());
migratedSerializedValueOutput.clear();
iterator.next();
@@ -1710,25 +1692,16 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
// TODO we implement the simple way of supporting the current functionality, mimicking keyed state
// because this should be reworked in FLINK-9376 and then we should have a common algorithm over
// StateMetaInfoSnapshot that avoids this code duplication.
- StateMetaInfoSnapshot restoredMetaInfoSnapshot = restoredKvStateMetaInfos.get(stateName);
-
- Preconditions.checkState(
- restoredMetaInfoSnapshot != null,
- "Requested to check compatibility of a restored RegisteredKeyedBackendStateMetaInfo," +
- " but its corresponding restored snapshot cannot be found.");
- StateMetaInfoSnapshot.CommonSerializerKeys serializerKey =
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER;
-
- TypeSerializer<?> metaInfoTypeSerializer = restoredMetaInfoSnapshot.restoreTypeSerializer(serializerKey);
+ @SuppressWarnings("unchecked")
+ RegisteredPriorityQueueStateBackendMetaInfo<T> castedMetaInfo =
+ (RegisteredPriorityQueueStateBackendMetaInfo<T>) metaInfoTuple.f1;
- if (metaInfoTypeSerializer != byteOrderedElementSerializer) {
- @SuppressWarnings("unchecked")
- TypeSerializerSnapshot<T> serializerSnapshot = Preconditions.checkNotNull(
- (TypeSerializerSnapshot<T>) restoredMetaInfoSnapshot.getTypeSerializerConfigSnapshot(serializerKey));
+ TypeSerializer<T> previousElementSerializer = castedMetaInfo.getPreviousElementSerializer();
+ if (previousElementSerializer != byteOrderedElementSerializer) {
TypeSerializerSchemaCompatibility<T> compatibilityResult =
- serializerSnapshot.resolveSchemaCompatibility(byteOrderedElementSerializer);
+ castedMetaInfo.updateElementSerializer(byteOrderedElementSerializer);
// Since priority queue elements are written into RocksDB
// as keys prefixed with the key group and namespace, we do not support
[flink] 07/15: [hotfix] [tests] Remove unused enum from
StateBackendTestBase
Posted by tz...@apache.org.
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 735e2f8d0d0a3f890b7a215f4c6657461bb732df
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sun Dec 9 16:08:07 2018 +0800
[hotfix] [tests] Remove unused enum from StateBackendTestBase
---
.../java/org/apache/flink/runtime/state/StateBackendTestBase.java | 8 --------
1 file changed, 8 deletions(-)
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 8708613..a9bd1f6 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
@@ -149,14 +149,6 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
@Rule
public final ExpectedException expectedException = ExpectedException.none();
- /**
- * The serialization timeliness behaviour of the state backend under test.
- */
- public enum BackendSerializationTimeliness {
- ON_ACCESS,
- ON_CHECKPOINTS
- }
-
// lazily initialized stream storage
private CheckpointStorageLocation checkpointStorageLocation;
[flink] 13/15: [hotfix] [docs] Correct the parameter in Operators
Overview doc
Posted by tz...@apache.org.
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 0078a7c5eb62725ff74bd2411853125096e52d0b
Author: KarmaGYZ <ka...@gmail.com>
AuthorDate: Mon Dec 3 19:09:28 2018 +0800
[hotfix] [docs] Correct the parameter in Operators Overview doc
This closes #7219.
---
docs/dev/stream/operators/index.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/dev/stream/operators/index.md b/docs/dev/stream/operators/index.md
index 422dbbf..b89fa89 100644
--- a/docs/dev/stream/operators/index.md
+++ b/docs/dev/stream/operators/index.md
@@ -434,14 +434,14 @@ IterativeStream<Long> iteration = initialStream.iterate();
DataStream<Long> iterationBody = iteration.map (/*do something*/);
DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
@Override
- public boolean filter(Integer value) throws Exception {
+ public boolean filter(Long value) throws Exception {
return value > 0;
}
});
iteration.closeWith(feedback);
DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
@Override
- public boolean filter(Integer value) throws Exception {
+ public boolean filter(Long value) throws Exception {
return value <= 0;
}
});