Github user shixiaogang commented on a diff in the pull request: https://github.com/apache/flink/pull/3336#discussion_r102128355 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java --- @@ -93,6 +95,18 @@ public DefaultKeyedStateStore(KeyedStateBackend<?> keyedStateBackend, ExecutionC } } + @Override + public <UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties) { + requireNonNull(stateProperties, "The state properties must not be null"); + try { + stateProperties.initializeSerializerUnlessSet(executionConfig); + MapState<UK, UV> originalState = getPartitionedState(stateProperties); + return new UserFacingMapState<>(originalState); + } catch (Exception e) { + throw new RuntimeException("Error while getting state", e); --- End diff -- Currently, `KeyedStateStore#getState()` does not throw exception in its declaration. `RuntimeException` is the only exception that can be thrown. Since the modification to the interface will affect user code (users will have to deal with thrown exceptions), I am not sure it's okay to modify the function declaration in `KeyedStateStore`.
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---