Repository: flink
Updated Branches:
  refs/heads/master 24c30878e -> 9e139a72b


[FLINK-8423] OperatorChain#pushToOperator catch block may fail with NPE

This closes #5447.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9e139a72
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9e139a72
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9e139a72

Branch: refs/heads/master
Commit: 9e139a72ba45f2dd820bd3b9ecdf8428588666fd
Parents: 24c3087
Author: zhangminglei <zml13856086...@163.com>
Authored: Tue Feb 13 10:33:04 2018 +0800
Committer: zentol <ches...@apache.org>
Committed: Tue Feb 13 09:51:57 2018 +0100

----------------------------------------------------------------------
 .../streaming/runtime/tasks/OperatorChain.java  | 25 +++++++++++---------
 1 file changed, 14 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/9e139a72/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
----------------------------------------------------------------------
diff --git 
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
 
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
index fdeea17..f3c7293 100644
--- 
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
+++ 
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
@@ -591,17 +591,20 @@ public class OperatorChain<OUT, OP extends 
StreamOperator<OUT>> implements Strea
                                operator.setKeyContextElement1(copy);
                                operator.processElement(copy);
                        } catch (ClassCastException e) {
-                               // Enrich error message
-                               ClassCastException replace = new 
ClassCastException(
-                                       String.format(
-                                               "%s. Failed to push OutputTag 
with id '%s' to operator. " +
-                                               "This can occur when multiple 
OutputTags with different types " +
-                                               "but identical names are being 
used.",
-                                               e.getMessage(),
-                                               outputTag.getId()));
-
-                               throw new 
ExceptionInChainedOperatorException(replace);
-
+                               if (outputTag != null) {
+                                       // Enrich error message
+                                       ClassCastException replace = new 
ClassCastException(
+                                               String.format(
+                                                       "%s. Failed to push 
OutputTag with id '%s' to operator. " +
+                                                               "This can occur 
when multiple OutputTags with different types " +
+                                                               "but identical 
names are being used.",
+                                                       e.getMessage(),
+                                                       outputTag.getId()));
+
+                                       throw new 
ExceptionInChainedOperatorException(replace);
+                               } else {
+                                       throw new 
ExceptionInChainedOperatorException(e);
+                               }
                        } catch (Exception e) {
                                throw new 
ExceptionInChainedOperatorException(e);
                        }

Reply via email to