Repository: flink Updated Branches: refs/heads/release-1.1 4526005d2 -> 33df945fe
[FLINK-4631] Prevent NPE in TwoInputStreamTask Check that the input processor has been created before cleaning it up. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/33df945f Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/33df945f Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/33df945f Branch: refs/heads/release-1.1 Commit: 33df945feee6019caa08f71e47d91a36cb5c54bd Parents: 4526005 Author: Till Rohrmann <trohrm...@apache.org> Authored: Wed Dec 7 17:28:20 2016 +0100 Committer: Till Rohrmann <trohrm...@apache.org> Committed: Wed Dec 7 17:29:58 2016 +0100 ---------------------------------------------------------------------- .../apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/33df945f/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java index 8718b88..0f2f3d8 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java @@ -95,7 +95,9 @@ public class TwoInputStreamTask<IN1, IN2, OUT> extends StreamTask<OUT, TwoInputS @Override protected void cleanup() throws Exception { - inputProcessor.cleanup(); + if (inputProcessor != null) { + inputProcessor.cleanup(); + } } @Override
