[beam] branch master updated: [BEAM-7091] fix NPE in DoFnOperator#dispose

[mxm]

This is an automated email from the ASF dual-hosted git repository.

mxm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new bbfe8a7  [BEAM-7091] fix NPE in DoFnOperator#dispose
     new 3682d0c  Merge pull request #8327: [BEAM-7091] Fix NPE in 
DoFnOperator#dispose
bbfe8a7 is described below

commit bbfe8a75c1905fe8a1f04bfc4703799f01609943
Author: Jan Lukavsky <je...@seznam.cz>
AuthorDate: Wed Apr 17 10:56:40 2019 +0200

    [BEAM-7091] fix NPE in DoFnOperator#dispose
---
 .../runners/flink/translation/wrappers/streaming/DoFnOperator.java   | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git 
a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
 
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
index dfed839..aa1f180 100644
--- 
a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
+++ 
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
@@ -29,6 +29,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -434,9 +435,9 @@ public class DoFnOperator<InputT, OutputT> extends 
AbstractStreamOperator<Window
   @Override
   public void dispose() throws Exception {
     try {
-      checkFinishBundleTimer.cancel(true);
+      Optional.ofNullable(checkFinishBundleTimer).ifPresent(timer -> 
timer.cancel(true));
       FlinkClassloading.deleteStaticCaches();
-      doFnInvoker.invokeTeardown();
+      Optional.ofNullable(doFnInvoker).ifPresent(DoFnInvoker::invokeTeardown);
     } finally {
       // This releases all task's resources. We need to call this last
       // to ensure that state, timers, or output buffers can still be