[FLINK-38486] Harden shutdown of system UDFs

If a resource is lazily created in open, we can only close after checking for null. Otherwise a failure during initialization will trigger secondary failures.

Author: AHeise

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java

@@ -210,7 +210,9 @@ public void writeRecord(T record) {
     @Override
     public void close() throws IOException {
         try {
-            writer.close();
+            if (writer != null) {
+                writer.close();
+            }
         } catch (Exception e) {
             throw new TableException("Exception in close", e);
         }

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java

@@ -471,8 +471,10 @@ public void writeRecord(RowData record) throws IOException {
 
             @Override
             public void close() throws IOException {
-                this.output.flush();
-                this.output.close();
+                if (output != null) {
+                    this.output.flush();
+                    this.output.close();
+                }
             }
         };
     }

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/batch/compact/BatchFileWriter.java

@@ -126,7 +126,9 @@ public void endInput() throws Exception {}
     public void close() throws Exception {
         try {
             staticPartitions.clear();
-            writer.close();
+            if (writer != null) {
+                writer.close();
+            }
         } catch (Exception e) {
             throw new TableException("Exception in close", e);
         }

flink-python/src/main/java/org/apache/flink/table/runtime/arrow/serializers/ArrowSerializer.java

@@ -113,10 +113,18 @@ public void write(RowData element) {
     }
 
     public void close() throws Exception {
-        arrowStreamWriter.end();
-        arrowStreamReader.close();
-        rootWriter.close();
-        allocator.close();
+        if (arrowStreamWriter != null) {
+            arrowStreamWriter.end();
+        }
+        if (arrowStreamReader != null) {
+            arrowStreamReader.close();
+        }
+        if (rootWriter != null) {
+            rootWriter.close();
+        }
+        if (allocator != null) {
+            allocator.close();
+        }
     }
 
     /** Creates an {@link ArrowWriter}. */

flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/aggregate/arrow/batch/BatchArrowPythonGroupWindowAggregateFunctionOperator.java

@@ -129,7 +129,9 @@ public void open() throws Exception {
     @Override
     public void close() throws Exception {
         super.close();
-        windowsGrouping.close();
+        if (windowsGrouping != null) {
+            windowsGrouping.close();
+        }
     }
 
     @Override

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/DefaultExpressionEvaluator.java

@@ -76,7 +76,9 @@ public MethodHandle open(FunctionContext context) {
     @Override
     public void close() {
         try {
-            instance.close();
+            if (instance != null) {
+                instance.close();
+            }
         } catch (Exception e) {
             throw new TableException(
                     String.format(

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/BufferDataOverWindowOperator.java

@@ -138,6 +138,8 @@ private void processCurrentData() throws Exception {
     @Override
     public void close() throws Exception {
         super.close();
-        this.currentData.close();
+        if (this.currentData != null) {
+            this.currentData.close();
+        }
     }
 }