Repository: flink Updated Branches: refs/heads/master 4410c04a6 -> 13150a4ba
[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/13150a4b Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/13150a4b Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/13150a4b Branch: refs/heads/master Commit: 13150a4ba26127b9ee2035fd3509b57bc3f7aa61 Parents: 4410c04 Author: Till Rohrmann <trohrm...@apache.org> Authored: Wed Dec 7 17:24:37 2016 +0100 Committer: Till Rohrmann <trohrm...@apache.org> Committed: Wed Dec 7 17:24:37 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/13150a4b/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 545b95b..233e9f1 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 @@ -91,7 +91,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