Zakelly commented on code in PR #26651:
URL: https://github.com/apache/flink/pull/26651#discussion_r2234476547
##########
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:
Thanks for the investigation and update! Current implementation LGTM as it
avoids the recalculation of key group as well as the concurrency issue.
--
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]
