hejufang commented on code in PR #26651:
URL: https://github.com/apache/flink/pull/26651#discussion_r2160799068
##########
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:
I haven't found any concurrency issues in unit tests or my testing jobs yet.
Additionally, I debug the implementation of StateTable.put and implementation
of CopyOnWriteStateMap.put, it locates the existing StateMapEntry and modifies
its state data without creating a new StateMapEntry or updating modCount.
Although I haven't identified concurrency issues so far, to avoid potential
problems, I still changed to using the transform method to update entries. WDYT?
--
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]
