(flink) branch master updated: [FLINK-35786] Fix NPE BlobServer / shutdownHook

[roman]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 050767cda9d [FLINK-35786] Fix NPE BlobServer / shutdownHook
050767cda9d is described below

commit 050767cda9de4f41748b0169b28679a87b5c5a11
Author: Roman Khachatryan <khachatryan.ro...@gmail.com>
AuthorDate: Mon Jul 8 15:54:58 2024 +0200

    [FLINK-35786] Fix NPE BlobServer / shutdownHook
---
 .../org/apache/flink/runtime/blob/BlobServer.java  | 25 ++++++++++++++--------
 1 file changed, 16 insertions(+), 9 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java 
b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
index 2a3c53ab1c1..56df2fb05cd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
@@ -97,7 +97,8 @@ public class BlobServer extends Thread
     private final AtomicLong tempFileCounter = new AtomicLong(0);
 
     /** The server socket listening for incoming connections. */
-    private final ServerSocket serverSocket;
+    // can be null if BlobServer is shut down before constructor completion
+    @Nullable private final ServerSocket serverSocket;
 
     /** Blob Server configuration. */
     private final Configuration blobServiceConfiguration;
@@ -354,10 +355,12 @@ public class BlobServer extends Thread
         if (shutdownRequested.compareAndSet(false, true)) {
             Exception exception = null;
 
-            try {
-                this.serverSocket.close();
-            } catch (IOException ioe) {
-                exception = ioe;
+            if (serverSocket != null) {
+                try {
+                    this.serverSocket.close();
+                } catch (IOException ioe) {
+                    exception = ioe;
+                }
             }
 
             // wake the thread up, in case it is waiting on some operation
@@ -394,10 +397,14 @@ public class BlobServer extends Thread
             ShutdownHookUtil.removeShutdownHook(shutdownHook, 
getClass().getSimpleName(), LOG);
 
             if (LOG.isInfoEnabled()) {
-                LOG.info(
-                        "Stopped BLOB server at {}:{}",
-                        serverSocket.getInetAddress().getHostAddress(),
-                        getPort());
+                if (serverSocket != null) {
+                    LOG.info(
+                            "Stopped BLOB server at {}:{}",
+                            serverSocket.getInetAddress().getHostAddress(),
+                            getPort());
+                } else {
+                    LOG.info("Stopped BLOB server before initializing the 
socket");
+                }
             }
 
             ExceptionUtils.tryRethrowIOException(exception);