Zakelly commented on code in PR #26651:
URL: https://github.com/apache/flink/pull/26651#discussion_r2154978638
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java:
##########
@@ -212,7 +214,7 @@ private <N, V> StateTable<K, N, V> tryRegisterStateTable(
StateDescriptor<?, V> stateDesc,
@Nonnull StateSnapshotTransformFactory<V> snapshotTransformFactory,
boolean allowFutureMetadataUpdates)
- throws StateMigrationException {
+ throws Exception {
Review Comment:
I'd agree keep the original exception, as the `FlinkRuntimeException` is
uncheck one and there is no need to claim throwing it.
##########
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:
It there any concurrency issue when performing `stateTable.put()` over the
iteration?
If so, I'd suggest a new inner transform method to in-place update all the
value of inner entries
--
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]
