[
https://issues.apache.org/jira/browse/FLINK-8715?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16456488#comment-16456488
]
ASF GitHub Bot commented on FLINK-8715:
---------------------------------------
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/5885#discussion_r184696134
--- Diff:
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
---
@@ -1116,148 +1115,177 @@ private void
restoreKeyGroupsShardWithTemporaryHelperInstance(
//
------------------------------------------------------------------------
/**
- * Creates a column family handle for use with a k/v state. When
restoring from a snapshot
- * we don't restore the individual k/v states, just the global RocksDB
database and the
- * list of column families. When a k/v state is first requested we
check here whether we
- * already have a column family for that and return it or create a new
one if it doesn't exist.
+ * Registers a k/v state information, which includes its state id,
type, RocksDB column family handle, and serializers.
*
- * <p>This also checks whether the {@link StateDescriptor} for a state
matches the one
- * that we checkpointed, i.e. is already in the map of column families.
+ * When restoring from a snapshot, we don’t restore the individual k/v
states, just the global RocksDB database and
+ * the list of k/v state information. When a k/v state is first
requested we check here whether we
+ * already have a registered entry for that and return it (after some
necessary state compatibility checks)
+ * or create a new one if it does not exist.
*/
- @SuppressWarnings("rawtypes, unchecked")
- protected <N, S> ColumnFamilyHandle getColumnFamily(
- StateDescriptor<?, S> descriptor, TypeSerializer<N>
namespaceSerializer) throws IOException, StateMigrationException {
+ private Tuple2<ColumnFamilyHandle,
RegisteredKeyedBackendStateMetaInfo<?, ?>> tryRegisterKvStateInformation(
--- End diff --
This method is rarely invoked, the tuples are not immutable so giving a
defensive copy can also have its benefits an I feel like this is cleaner than
having casts all over the place. There might also be different ways to solve
the general problem, but this feels better to me than the initial version.
> RocksDB does not propagate reconfiguration of serializer to the states
> ----------------------------------------------------------------------
>
> Key: FLINK-8715
> URL: https://issues.apache.org/jira/browse/FLINK-8715
> Project: Flink
> Issue Type: Bug
> Components: State Backends, Checkpointing
> Affects Versions: 1.3.2
> Reporter: Arvid Heise
> Assignee: Tzu-Li (Gordon) Tai
> Priority: Blocker
> Fix For: 1.5.0
>
>
> Any changes to the serializer done in #ensureCompability are lost during the
> state creation.
> In particular,
> [https://github.com/apache/flink/blob/master/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java#L68]
> always uses a fresh copy of the StateDescriptor.
> An easy fix is to pass the reconfigured serializer as an additional parameter
> in
> [https://github.com/apache/flink/blob/master/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java#L1681]
> , which can be retrieved through the side-output of getColumnFamily
> {code:java}
> kvStateInformation.get(stateDesc.getName()).f1.getStateSerializer()
> {code}
> I encountered it in 1.3.2 but the code in the master seems unchanged (hence
> the pointer into master). I encountered it in ValueState, but I suspect the
> same issue can be observed for all kinds of RocksDB states.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
