hejufang commented on code in PR #26651:
URL: https://github.com/apache/flink/pull/26651#discussion_r2173697813
##########
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:
@Zakelly Thank you for your reply. I have reviewed the implementations of
StateTable and StateMap and conducted some tests. Here is my answer your
questions.
Will iterating over StateMap using an iterator and performing transform
throw a ConcurrentModificationException?
No, it won't. This is because the putEntry method locates an already
existing StateMapEntry instead of creating a new one. It returns before
updating modCount (see CopyOnWriteStateMap#putEntry at line 399), so when
transforming existing data, modCount does not change, and therefore no
ConcurrentModificationException is thrown.
Are there other concurrency issues?
Yes.
I verified that the following code will iterate over duplicate data:
```
for (int i = 0; i < keyGroupedStateMaps.length; i++) {
Iterator<StateEntry<K, N, S>> iterator =
keyGroupedStateMaps[i].iterator();
while (iterator.hasNext()) {
StateEntry<K, N, S> entry = iterator.next();
keyGroupedStateMaps[i].transform(
entry.getKey(), entry.getNamespace(), value, transformation);
}
}
```
This happens because when iterating directly over
CopyOnWriteStateMap.iterator(), calling transform internally invokes putEntry,
which triggers computeHashForOperationAndDoIncrementalRehash. This causes data
to be moved between two tables (primaryTable and incrementalRehashTable). If
the iterator is traversing primaryTable when transform triggers rehashing, some
data will move to incrementalRehashTable. The iterator may later visit these
entries in incrementalRehashTable, causing duplicate processing.
This issue is easy to reproduce by adding more data to StateMap in a unit
test. For instance, StateBackendMigrationTestBase.testKeyedValueStateUpgrade,
by adding the following code to insert more entries:
```
for (int i = 0; i < 10000; i++) {
backend.setCurrentKey(i);
valueState.update(new TestType("test" + i, i * 1000));
}
```
Running
HashMapStateBackendMigrationTest.testStateMigrationAfterChangingTTLFromDisablingToEnabling
then results in errors.
Why does using StateTable.iterator() not iterate duplicate entries?
Because StateTable.iterator() returns a Spliterator. Before traversal, the
Spliterator reads the original iterator (StateMap.iterator()) and preloads the
data into a cache
(java.util.stream.StreamSpliterators.AbstractWrappingSpliterator.buffer). This
effectively creates a snapshot of all entries. Therefore, even if subsequent
transforms cause data to move, the iterator will not visit duplicate entries.
Considering that recalculating the key group involves some performance
overhead, I optimized the code referencing StateTable.iterator() logic:
First,
read all data before executing any transform, then update the data:
```
for (StateMap<K, N, S> stateMap : keyGroupedStateMaps) {
List<StateEntry<K, N, S>> entries =
StreamSupport.stream(
Spliterators.spliteratorUnknownSize(stateMap.iterator(), 0),
false)
.collect(Collectors.toList());
for (StateEntry<K, N, S> entry : entries) {
stateMap.transform(entry.getKey(), entry.getNamespace(), value,
transformation);
}
}
```
Is there any other possible approach?
I think another possibility is to add a transformAll interface to StateMap,
which directly iterates over both primaryTable and incrementalRehashTable
inside CopyOnWriteStateMap. However, I'm unsure if this introduces other
potential issues, and it would also require adding implementations to
CopyOnWriteSkipListStateMap, which could be complex. So I prefer the approach
above which I have verified in my test jobs. What do you think?
--
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]
