commit a08141e58f2918d8659cf52f035d2660df4063f2
author Igal Shilman <igalshilman@gmail.com> Mon Sep 21 16:15:00 2020 +0200
committer Tzu-Li (Gordon) Tai <tzulitai@apache.org> Tue Sep 22 16:22:50 2020 +0800
tree 7f9333aa6140df3bc969f6cd3aab967644192b5e
parent 0d85a2c7b8bf6b1e0568d6e7ffcc8bbd0d5562ad

[FLINK-19329] Check for the existence of managed resources during dispose

This closes #158.

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java

diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
index 9cc0818..381a672 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
@@ -184,6 +184,12 @@
   }
 
   private void closeOrDispose() {
+    final List<ManagingResources> managingResources = this.managingResources;
+    if (managingResources == null) {
+      // dispose can be called before state initialization was completed (for example a failure
+      // during initialization).
+      return;
+    }
     for (ManagingResources withResources : managingResources) {
       try {
         withResources.shutdown();