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 2019/04/01 04:26:19 UTC
[flink] branch release-1.8 updated: [FLINK-12064] [core,
State Backends] RocksDBKeyedStateBackend snapshots uses incorrect
key serializer if reconfigure happens during restore
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch release-1.8
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.8 by this push:
new 9795457 [FLINK-12064] [core, State Backends] RocksDBKeyedStateBackend snapshots uses incorrect key serializer if reconfigure happens during restore
9795457 is described below
commit 9795457a0228e055b82049977ac84d5bc70e6a23
Author: Yu Li <li...@apache.org>
AuthorDate: Fri Mar 29 18:24:01 2019 +0800
[FLINK-12064] [core, State Backends] RocksDBKeyedStateBackend snapshots uses incorrect key serializer if reconfigure happens during restore
This closes #8076.
---
.../state/AbstractKeyedStateBackendBuilder.java | 2 --
.../state/StateBackendMigrationTestBase.java | 25 ++++++++++++++++++----
.../ttl/mock/MockKeyedStateBackendBuilder.java | 2 +-
.../state/RocksDBKeyedStateBackendBuilder.java | 4 ++--
4 files changed, 24 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackendBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackendBuilder.java
index addb549..d7573fd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackendBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackendBuilder.java
@@ -34,7 +34,6 @@ import java.util.Collection;
public abstract class AbstractKeyedStateBackendBuilder<K>
implements StateBackendBuilder<AbstractKeyedStateBackend, BackendBuildingException> {
protected final TaskKvStateRegistry kvStateRegistry;
- protected final TypeSerializer<K> keySerializer;
protected final StateSerializerProvider<K> keySerializerProvider;
protected final ClassLoader userCodeClassLoader;
protected final int numberOfKeyGroups;
@@ -57,7 +56,6 @@ public abstract class AbstractKeyedStateBackendBuilder<K>
StreamCompressionDecorator keyGroupCompressionDecorator,
CloseableRegistry cancelStreamRegistry) {
this.kvStateRegistry = kvStateRegistry;
- this.keySerializer = keySerializer;
this.keySerializerProvider = StateSerializerProvider.fromNewRegisteredSerializer(keySerializer);
this.userCodeClassLoader = userCodeClassLoader;
this.numberOfKeyGroups = numberOfKeyGroups;
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 3ba8876..8c48398 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
@@ -157,8 +157,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
CustomVoidNamespaceSerializer.INSTANCE,
newAccessDescriptorAfterRestore);
- snapshot.discardState();
-
// make sure that reading and writing each key state works with the new serializer
backend.setCurrentKey(1);
Assert.assertEquals(new TestType("foo", 1456), valueState.value());
@@ -171,6 +169,12 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend.setCurrentKey(3);
Assert.assertEquals(new TestType("hello", 189), valueState.value());
valueState.update(new TestType("newValue3", 444));
+
+ // do another snapshot to verify the snapshot logic after migration
+ snapshot = runSnapshot(
+ backend.snapshot(2L, 3L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
+ snapshot.discardState();
} finally {
backend.dispose();
}
@@ -266,8 +270,6 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
CustomVoidNamespaceSerializer.INSTANCE,
newAccessDescriptorAfterRestore);
- snapshot.discardState();
-
// make sure that reading and writing each key state works with the new serializer
backend.setCurrentKey(1);
Iterator<TestType> iterable1 = listState.get().iterator();
@@ -289,6 +291,13 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
Assert.assertEquals(new TestType("key-3", 2), iterable3.next());
Assert.assertFalse(iterable3.hasNext());
listState.add(new TestType("new-key-3", 777));
+
+ // do another snapshot to verify the snapshot logic after migration
+ snapshot = runSnapshot(
+ backend.snapshot(2L, 3L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
+ snapshot.discardState();
+
} finally {
backend.dispose();
}
@@ -405,6 +414,10 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend.setCurrentKey(new TestType("bar", 456));
Assert.assertEquals(5, valueState.value().intValue());
+ // do another snapshot to verify the snapshot logic after migration
+ snapshot = runSnapshot(
+ backend.snapshot(2L, 3L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
snapshot.discardState();
} finally {
backend.dispose();
@@ -493,6 +506,10 @@ public abstract class StateBackendMigrationTestBase<B extends AbstractStateBacke
backend.setCurrentKey(5);
Assert.assertEquals(50, valueState.value().intValue());
+ // do another snapshot to verify the snapshot logic after migration
+ snapshot = runSnapshot(
+ backend.snapshot(2L, 3L, streamFactory, CheckpointOptions.forCheckpointWithDefaultLocation()),
+ sharedStateRegistry);
snapshot.discardState();
} finally {
backend.dispose();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackendBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackendBuilder.java
index 5706fb0..efef923 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackendBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackendBuilder.java
@@ -72,7 +72,7 @@ public class MockKeyedStateBackendBuilder<K> extends AbstractKeyedStateBackendBu
restoreOperation.restore();
return new MockKeyedStateBackend<>(
kvStateRegistry,
- keySerializer,
+ keySerializerProvider.currentSchemaSerializer(),
userCodeClassLoader,
numberOfKeyGroups,
keyGroupRange,
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
index aa6845e..9b98e6e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackendBuilder.java
@@ -425,7 +425,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
RocksDBSnapshotStrategyBase<K> savepointSnapshotStrategy = new RocksFullSnapshotStrategy<>(
db,
rocksDBResourceGuard,
- keySerializer,
+ keySerializerProvider.currentSchemaSerializer(),
kvStateInformation,
keyGroupRange,
keyGroupPrefixBytes,
@@ -438,7 +438,7 @@ public class RocksDBKeyedStateBackendBuilder<K> extends AbstractKeyedStateBacken
checkpointSnapshotStrategy = new RocksIncrementalSnapshotStrategy<>(
db,
rocksDBResourceGuard,
- keySerializer,
+ keySerializerProvider.currentSchemaSerializer(),
kvStateInformation,
keyGroupRange,
keyGroupPrefixBytes,