You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by StefanRRichter <gi...@git.apache.org> on 2016/12/08 16:07:21 UTC
[GitHub] flink pull request #2969: [FLINK-5289] Meaningful exception when using value...
GitHub user StefanRRichter opened a pull request:
https://github.com/apache/flink/pull/2969
[FLINK-5289] Meaningful exception when using value state on non-keyed\u2026
This PR fixes [FLINK-5289] and introduces a meaningful exception when registering a keyed state on non-keyed stream.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/StefanRRichter/flink register-state-NPE
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2969.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2969
----
commit fba7f61ee6ff7155ee166ec01134f81f0b7f4457
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-12-08T14:07:58Z
[FLINK-5289] Meaningful exception when using value state on non-keyed stream
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2969: [FLINK-5289] Meaningful exception when using value...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2969
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2969: [FLINK-5289] Meaningful exception when using value state ...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2969
+1 looks good. Will put this into my merge pipeline
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2969: [FLINK-5289] Meaningful exception when using value...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2969#discussion_r91703759
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java ---
@@ -106,17 +109,45 @@ public boolean hasBroadcastVariable(String name) {
@Override
public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
+ }
}
@Override
public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getListState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getListState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
+ }
}
@Override
public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getReducingState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getReducingState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
+ }
+ }
+
+ private KeyedStateStore checkPreconditionsAndGetKeyedStateStore(StateDescriptor<?, ?> stateDescriptor) {
+ Preconditions.checkNotNull(stateDescriptor, "The state properties must not be null");
+ KeyedStateStore keyedStateStore = operator.getKeyedStateStore();
+ Preconditions.checkNotNull(keyedStateStore, "Keyed state store is null. This can only be called after a keyBy.");
--- End diff --
How about changing this message to
`Keyed state can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2969: [FLINK-5289] Meaningful exception when using value...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2969#discussion_r91703632
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java ---
@@ -106,17 +109,45 @@ public boolean hasBroadcastVariable(String name) {
@Override
public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
+ }
}
@Override
public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getListState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getListState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
+ }
}
@Override
public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
- return operator.getKeyedStateStore().getReducingState(stateProperties);
+ KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
+
+ try {
+ stateProperties.initializeSerializerUnlessSet(getExecutionConfig());
+ return keyedStateStore.getReducingState(stateProperties);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state", e);
--- End diff --
Do we need to re-wrap the exceptions here? Can we simply let the original exception bubble up?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2969: [FLINK-5289] Meaningful exception when using value state ...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/2969
CC @twalthr
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2969: [FLINK-5289] Meaningful exception when using value state ...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/2969
Updated w.r.t. comments.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---