Zakelly commented on code in PR #26651:
URL: https://github.com/apache/flink/pull/26651#discussion_r2162949352
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java:
##########
@@ -299,6 +296,52 @@ private <N, V> StateTable<K, N, V> tryRegisterStateTable(
return stateTable;
}
+ @SuppressWarnings("unchecked")
+ private <V, N> void migrateTtlAwareStateValues(
+ StateDescriptor<?, V> stateDesc,
+ TypeSerializer<V> previousSerializer,
+ TypeSerializer<V> currentSerializer)
+ throws Exception {
+ final StateTable<K, N, V> stateTable =
+ (StateTable<K, N, V>)
registeredKVStates.get(stateDesc.getName());
+ final Iterator<StateEntry<K, N, V>> iterator = stateTable.iterator();
+
+ LOG.info(
+ "Performing state migration for state {} because the state
serializer's ttl"
+ + " config has been changed from {} to {}.",
+ stateDesc,
+ TtlAwareSerializer.isSerializerTtlEnabled(previousSerializer),
+ TtlAwareSerializer.isSerializerTtlEnabled(currentSerializer));
+
+ // we need to get an actual state instance because migration is
different
+ // for different state types. For example, ListState needs to deal with
+ // individual elements
+ StateCreateFactory stateCreateFactory =
STATE_CREATE_FACTORIES.get(stateDesc.getType());
+ if (stateCreateFactory == null) {
+ throw new
FlinkRuntimeException(stateNotSupportedMessage(stateDesc));
+ }
+ State state =
+ stateCreateFactory.createState(stateDesc, stateTable,
stateTable.keySerializer);
+ if (!(state instanceof AbstractHeapState)) {
+ throw new FlinkRuntimeException(
+ "State should be an AbstractRocksDBState but is " + state);
+ }
+ AbstractHeapState<K, N, V> heapState = (AbstractHeapState<K, N, V>)
state;
+ TtlAwareSerializer<V, ?> currentTtlAwareSerializer =
+ (TtlAwareSerializer<V, ?>)
+
TtlAwareSerializer.wrapTtlAwareSerializer(currentSerializer);
+
+ while (iterator.hasNext()) {
+ final StateEntry<K, N, V> entry = iterator.next();
+ stateTable.put(
Review Comment:
@hejufang I'm afraid the current implementation does not avoid the potential
concurrency issue and will achieve low performance.
To be more specific, the `CopyOnWriteStateMap$StateEntryIterator` will check
the `expectedModCount` in each `next()` call, but the `modCount` will update
when `putEntry()` during `transform()`. So will a
`ConcurrentModificationException` thrown? Correct me if I'm wrong.
In current `transformAll` implementation, we do entry iteration and then
invoke `transform`, the `getMapForKeyGroup` will calculate the key group again
and get the state map, which is redundant since the entry is iterated from the
corresponding map. I'm worried about the performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
