[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

@@ -124,7 +124,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-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java

@@ -100,9 +100,11 @@ public void open(MetricConfig metricConfig) {
 
     @Override
     public void close() {
-        exporter.flush();
-        lastResult.join(1, TimeUnit.MINUTES);
-        exporter.close();
+        if (exporter != null) {
+            exporter.flush();
+            lastResult.join(1, TimeUnit.MINUTES);
+            exporter.close();
+        }
     }
 
     @Override

flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/traces/otel/OpenTelemetryTraceReporter.java

@@ -70,10 +70,14 @@ public void open(MetricConfig metricConfig) {
 
     @Override
     public void close() {
-        spanProcessor.forceFlush();
-        spanProcessor.close();
-        spanExporter.flush();
-        spanExporter.close();
+        if (spanProcessor != null) {
+            spanProcessor.forceFlush();
+            spanProcessor.close();
+        }
+        if (spanExporter != null) {
+            spanExporter.flush();
+            spanExporter.close();
+        }
     }
 
     private void notifyOfAddedSpanInternal(Span span, io.opentelemetry.api.trace.Span parent) {

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

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