Hi! In case where new state type is incompatible with old one from savepoint, we get IllegalStateException via org.apache.flink.util.Preconditions.checkState from checkStateMetaInfo: https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java#L220
For example, if an updated app uses reducing state but restoring from aggregating state, we can get the following exception: java.lang.IllegalStateException: Incompatible key/value state types. Was [AGGREGATING], registered with [REDUCING]. at org.apache.flink.util.Preconditions.checkState(Preconditions.java:195) at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.checkStateMetaInfo(RegisteredKeyValueStateBackendMetaInfo.java:216) at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.updateRestoredStateMetaInfo(RocksDBKeyedStateBackend.java:520) at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.tryRegisterKvStateInformation(RocksDBKeyedStateBackend.java:475) at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.createInternalState(RocksDBKeyedStateBackend.java:613) at org.apache.flink.runtime.state.KeyedStateFactory.createInternalState(KeyedStateFactory.java:47) at org.apache.flink.runtime.state.ttl.TtlStateFactory.createStateAndWrapWithTtlIfEnabled(TtlStateFactory.java:72) at org.apache.flink.runtime.state.AbstractKeyedStateBackend.getOrCreateKeyedState(AbstractKeyedStateBackend.java:286) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.getOrCreateKeyedState(AbstractStreamOperator.java:568) at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.open(WindowOperator.java:240) at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:438) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:298) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:714) at java.lang.Thread.run(Thread.java:748) The topic of this email is not about how to solve this issue, rather discussing about different type of exception to consistently inform users of incompatible state. I think StateMigrationException seems reasonable for this, as other types of state incompatibility is notified by that exception – key/namespace/value. Although the existing use of StateMigrationException is mostly from compatibility check from type serializer snapshot, its semantic seems to notify developers of state incompatibility in general. The type mismatch from state descriptor (like aggregating to reducing) seems more like upfront/fast check of incompatibility, so it looks better using StateMigrationException rather than too general IllegalStateException. A bit of context behind this thought is we internally consider IllegalStateException from flink core code base as sort of runtime state violation (e.g., internal state is inconsistent), so we generally expect this to be not from user’s mistake. Apparently, the above incompatible type is frequent mistakes from developers when restoring from savepoint. We would like to narrow such a specific type of user error down to a certain type of exception, not being thrown as general java exception. I would like to hear how you think about changing the exception type to StateMigrationException in checkStateMetaInfo. If it’s not reasonable, I wonder why and what alternative would be considered (e.g., another or new type). Thanks, Hwanju