[FLINK-39330][runtime] Make CollectSinkFunction teardown null-safe to fix flaky FileSourceTextLinesITCase JM failover

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java

@@ -300,11 +300,16 @@ public void invoke(IN value, Context context) throws Exception {
 
     @Override
     public void close() throws Exception {
-        serverThread.close();
-        serverThread.join();
+        if (serverThread != null) {
+            serverThread.close();
+            serverThread.join();
+        }
     }
 
     public void accumulateFinalResults() throws Exception {
+        if (bufferLock == null) {
+            return;
+        }
         bufferLock.lock();
         try {
             // put results not consumed by the client into the accumulator

flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java

@@ -34,6 +34,7 @@
 import java.util.Collections;
 import java.util.List;
 
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link CollectSinkFunction}. */
@@ -54,6 +55,19 @@ void after() throws Exception {
         functionWrapper.closeWrapper();
     }
 
+    @Test
+    void testCloseBeforeOpenDoesNotThrow() {
+        CollectSinkFunction<Integer> function =
+                new CollectSinkFunction<>(
+                        serializer, 12, CollectSinkFunctionTestWrapper.ACCUMULATOR_NAME);
+        assertThatCode(
+                        () -> {
+                            function.accumulateFinalResults();
+                            function.close();
+                        })
+                .doesNotThrowAnyException();
+    }
+
     @Test
     void testIncreasingToken() throws Exception {
         functionWrapper.openFunction();