[fix](fe) Prevent port conflicts during mocked FE startup

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -67,6 +67,8 @@
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.lang.management.ManagementFactory;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.OverlappingFileLockException;
@@ -143,7 +145,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
         CommandLine commandLine = parseArgs(args);
         CommandLineOptions cmdLineOpts = buildCommandLineOptions(commandLine);
 
-        try {
+        try (ServerSocket rpcListener = options.useEphemeralRpcPort ? new ServerSocket() : null) {
             // init config
             Config config = new Config();
             config.init(dorisHomeDir + "/conf/fe.conf");
@@ -231,8 +233,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
 
             FrontendOptions.init();
 
-            // check all port
-            checkAllPorts();
+            checkAllPorts(options);
 
             if (Config.enable_bdbje_debug_mode) {
                 // Start in BDB Debug mode
@@ -256,6 +257,14 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
                 }
                 Env.getCurrentEnv().setClusterSnapshotFile(clusterSnapshotPath);
             }
+            if (rpcListener != null) {
+                rpcListener.setReuseAddress(true);
+                rpcListener.bind(new InetSocketAddress(FrontendOptions.isBindIPV6() ? "::0" : "0.0.0.0", 0),
+                        Config.thrift_backlog_num);
+                // Env.initialize() advertises this port to the backends.
+                Config.rpc_port = rpcListener.getLocalPort();
+            }
+
             // init catalog and wait it be ready
             Env.getCurrentEnv().initialize(args);
             Env.getCurrentEnv().waitForReady();
@@ -268,7 +277,9 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             // 1. HttpServer for HTTP Server
             // 2. FE thrift server starter
             // 3. QeService for MySQL Server
-            thriftServerStarter = FeServerStarterFactory.createThriftServerStarter(Config.rpc_port);
+            thriftServerStarter = rpcListener == null
+                    ? FeServerStarterFactory.createThriftServerStarter(Config.rpc_port)
+                    : FeServerStarterFactory.createThriftServerStarter(rpcListener);
             thriftServerStarter.start();
 
             if (options.enableHttpServer) {
@@ -319,33 +330,36 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
         }
     }
 
-    private static void checkAllPorts() throws IOException {
+    static void checkAllPorts(StartupOptions options) throws IOException {
         if (Config.enable_check_compatibility_mode) {
             // The compatibility mode does not need to listen ports.
             return;
         }
 
-        if (!NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.edit_log_port,
-                "Edit log port", NetUtils.EDIT_LOG_PORT_SUGGESTION)) {
+        // LocalJournal writes to disk and does not open a replication listener.
+        if ("bdb".equalsIgnoreCase(Config.edit_log_type)
+                && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.edit_log_port,
+                        "Edit log port", NetUtils.EDIT_LOG_PORT_SUGGESTION)) {
             throw new IOException("port " + Config.edit_log_port + " already in use");
         }
-        if (!NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.http_port,
-                "Http port", NetUtils.HTTP_PORT_SUGGESTION)) {
+        if (options.enableHttpServer && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(),
+                Config.http_port, "Http port", NetUtils.HTTP_PORT_SUGGESTION)) {
             throw new IOException("port " + Config.http_port + " already in use");
         }
-        if (Config.enable_https && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(),
-                Config.https_port, "Https port", NetUtils.HTTPS_PORT_SUGGESTION)) {
+        if (options.enableHttpServer && Config.enable_https
+                && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.https_port,
+                        "Https port", NetUtils.HTTPS_PORT_SUGGESTION)) {
             throw new IOException("port " + Config.https_port + " already in use");
         }
-        if (!NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.query_port,
-                "Query port", NetUtils.QUERY_PORT_SUGGESTION)) {
+        if (options.enableQeService && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(),
+                Config.query_port, "Query port", NetUtils.QUERY_PORT_SUGGESTION)) {
             throw new IOException("port " + Config.query_port + " already in use");
         }
-        if (!NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.rpc_port,
-                "Rpc port", NetUtils.RPC_PORT_SUGGESTION)) {
+        if (!options.useEphemeralRpcPort && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(),
+                Config.rpc_port, "Rpc port", NetUtils.RPC_PORT_SUGGESTION)) {
             throw new IOException("port " + Config.rpc_port + " already in use");
         }
-        if (Config.arrow_flight_sql_port != -1
+        if (options.enableQeService && Config.arrow_flight_sql_port != -1
                 && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.arrow_flight_sql_port,
                 "Arrow Flight SQL port", NetUtils.ARROW_FLIGHT_SQL_SUGGESTION)) {
             throw new IOException("port " + Config.arrow_flight_sql_port + " already in use");
@@ -717,6 +731,7 @@ private static void setFuzzyForCatalog() {
     }
 
     public static class StartupOptions {
+        public boolean useEphemeralRpcPort;
         public boolean enableHttpServer = true;
         public boolean enableQeService = true;
     }

fe/fe-core/src/main/java/org/apache/doris/common/ThriftServer.java

@@ -95,23 +95,18 @@ public ThriftServerType getType() {
         return type;
     }
 
-    private void createSimpleServer() throws TTransportException {
-        TServerSocket.ServerSocketTransportArgs socketTransportArgs;
-
-        if (FrontendOptions.isBindIPV6()) {
-            socketTransportArgs = new TServerSocket.ServerSocketTransportArgs()
-                    .bindAddr(new InetSocketAddress("::0", port))
-                    .clientTimeout(Config.thrift_client_timeout_ms)
-                    .backlog(Config.thrift_backlog_num);
-        } else {
-            socketTransportArgs = new TServerSocket.ServerSocketTransportArgs()
-                    .bindAddr(new InetSocketAddress("0.0.0.0", port))
-                    .clientTimeout(Config.thrift_client_timeout_ms)
-                    .backlog(Config.thrift_backlog_num);
-        }
+    private TServerSocket createBlockingSocket(ServerSocket listener) throws TTransportException {
+        TServerSocket.ServerSocketTransportArgs args = new TServerSocket.ServerSocketTransportArgs()
+                .bindAddr(new InetSocketAddress(FrontendOptions.isBindIPV6() ? "::0" : "0.0.0.0", port))
+                .clientTimeout(Config.thrift_client_timeout_ms)
+                .backlog(Config.thrift_backlog_num)
+                .serverSocket(listener);
+        return new ImprovedTServerSocket(args);
+    }
 
-        TServer.Args args = new TServer.Args(new ImprovedTServerSocket(socketTransportArgs)).protocolFactory(
-                new TBinaryProtocol.Factory()).processor(processor);
+    private void createSimpleServer(ServerSocket listener) throws TTransportException {
+        TServer.Args args = new TServer.Args(createBlockingSocket(listener))
+                .protocolFactory(new TBinaryProtocol.Factory()).processor(processor);
         server = new TSimpleServer(args);
     }
 
@@ -140,22 +135,8 @@ private void createThreadedServer() throws TTransportException {
         server = new TThreadedSelectorServer(args);
     }
 
-    private void createThreadPoolServer() throws TTransportException {
-        TServerSocket.ServerSocketTransportArgs socketTransportArgs;
-
-        if (FrontendOptions.isBindIPV6()) {
-            socketTransportArgs = new TServerSocket.ServerSocketTransportArgs()
-                    .bindAddr(new InetSocketAddress("::0", port))
-                    .clientTimeout(Config.thrift_client_timeout_ms)
-                    .backlog(Config.thrift_backlog_num);
-        } else {
-            socketTransportArgs = new TServerSocket.ServerSocketTransportArgs()
-                    .bindAddr(new InetSocketAddress("0.0.0.0", port))
-                    .clientTimeout(Config.thrift_client_timeout_ms)
-                    .backlog(Config.thrift_backlog_num);
-        }
-
-        TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(new ImprovedTServerSocket(socketTransportArgs))
+    private void createThreadPoolServer(ServerSocket listener) throws TTransportException {
+        TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(createBlockingSocket(listener))
                 .protocolFactory(new TBinaryProtocol.Factory())
                 .processor(processor);
         ThreadPoolExecutor threadPoolExecutor = ThreadPoolManager.newDaemonCacheThreadPool(
@@ -165,16 +146,24 @@ private void createThreadPoolServer() throws TTransportException {
     }
 
     public void start() throws IOException {
+        start(null);
+    }
+
+    /** A blocking server can take ownership of a socket that is already listening. */
+    public void start(ServerSocket listener) throws IOException {
         try {
             switch (type) {
                 case SIMPLE:
-                    createSimpleServer();
+                    createSimpleServer(listener);
                     break;
                 case THREADED_SELECTOR:
+                    if (listener != null) {
+                        throw new IOException("THREADED_SELECTOR requires a nonblocking server socket");
+                    }
                     createThreadedServer();
                     break;
                 default:
-                    createThreadPoolServer();
+                    createThreadPoolServer(listener);
             }
         } catch (TTransportException ex) {
             LOG.warn("create thrift server failed.", ex);

fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterFactory.java

@@ -22,6 +22,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.io.IOException;
+import java.net.ServerSocket;
 import java.util.Iterator;
 import java.util.ServiceLoader;
 
@@ -39,6 +41,10 @@ public static ServerStarter createThriftServerStarter(int port) {
         return provider().createThriftServerStarter(port);
     }
 
+    public static ServerStarter createThriftServerStarter(ServerSocket listener) throws IOException {
+        return provider().createThriftServerStarter(listener);
+    }
+
     public static ServerStarter createFlightServerStarter(int port) {
         return provider().createFlightServerStarter(port);
     }

fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterProvider.java

@@ -19,12 +19,22 @@
 
 import org.apache.doris.qe.ConnectScheduler;
 
+import java.io.IOException;
+import java.net.ServerSocket;
+
 // FE startup splits at the starter provider boundary. OSS and TLS-enabled builds
 // return different protocol starters while DorisFE and QeService keep a stable
 // startup flow.
 public interface FeServerStarterProvider {
     ServerStarter createThriftServerStarter(int port);
 
+    /** Mock FEs supply a bound listener. Providers with their own transport bind at startup as usual. */
+    default ServerStarter createThriftServerStarter(ServerSocket listener) throws IOException {
+        int port = listener.getLocalPort();
+        listener.close();
+        return createThriftServerStarter(port);
+    }
+
     ServerStarter createMysqlServerStarter(int port, ConnectScheduler scheduler);
 
     ServerStarter createFlightServerStarter(int port);

fe/fe-core/src/main/java/org/apache/doris/tls/server/OssFeServerStarterProvider.java

@@ -32,12 +32,22 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.thrift.TProcessor;
 
+import java.io.IOException;
 import java.lang.reflect.Proxy;
+import java.net.ServerSocket;
 
 public class OssFeServerStarterProvider implements FeServerStarterProvider {
     @Override
     public ServerStarter createThriftServerStarter(int port) {
-        return new OssThriftServerStarter(port);
+        return new OssThriftServerStarter(port, null);
+    }
+
+    @Override
+    public ServerStarter createThriftServerStarter(ServerSocket listener) throws IOException {
+        if (ThriftServer.THREADED_SELECTOR.equalsIgnoreCase(Config.thrift_server_type)) {
+            return FeServerStarterProvider.super.createThriftServerStarter(listener);
+        }
+        return new OssThriftServerStarter(listener.getLocalPort(), listener);
     }
 
     @Override
@@ -53,10 +63,12 @@ public ServerStarter createFlightServerStarter(int port) {
     private static final class OssThriftServerStarter implements ServerStarter {
         private static final Logger LOG = LogManager.getLogger(OssThriftServerStarter.class);
         private final int port;
+        private final ServerSocket listener;
         private ThriftServer server;
 
-        private OssThriftServerStarter(int port) {
+        private OssThriftServerStarter(int port, ServerSocket listener) {
             this.port = port;
+            this.listener = listener;
         }
 
         @Override
@@ -89,7 +101,7 @@ public void start() throws Exception {
                     });
             TProcessor processor = new FrontendService.Processor<>(instance);
             server = new ThriftServer(port, processor);
-            server.start();
+            server.start(listener);
             LOG.info("FE thrift starter started on {}", port);
         }
 

fe/fe-core/src/test/java/org/apache/doris/DorisFEPortTest.java

@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris;
+
+import org.apache.doris.common.Config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+
+class DorisFEPortTest {
+    @Test
+    void occupiedPortsAreCheckedOnlyWhenTheirServicesStart() throws IOException {
+        String originalJournal = Config.edit_log_type;
+        int originalEditLogPort = Config.edit_log_port;
+        int originalQueryPort = Config.query_port;
+        try (ServerSocket editLog = new ServerSocket(0); ServerSocket query = new ServerSocket(0)) {
+            Config.edit_log_port = editLog.getLocalPort();
+            Config.query_port = query.getLocalPort();
+            Config.edit_log_type = "local";
+            DorisFE.StartupOptions options = new DorisFE.StartupOptions();
+            options.enableHttpServer = false;
+            options.enableQeService = false;
+            options.useEphemeralRpcPort = true;
+            DorisFE.checkAllPorts(options);
+
+            Config.edit_log_type = "bdb";
+            IOException editLogFailure = Assertions.assertThrows(IOException.class,
+                    () -> DorisFE.checkAllPorts(options));
+            Assertions.assertEquals("port " + editLog.getLocalPort() + " already in use", editLogFailure.getMessage());
+
+            Config.edit_log_type = "local";
+            options.enableQeService = true;
+            IOException queryFailure = Assertions.assertThrows(IOException.class,
+                    () -> DorisFE.checkAllPorts(options));
+            Assertions.assertEquals("port " + query.getLocalPort() + " already in use", queryFailure.getMessage());
+        } finally {
+            Config.edit_log_type = originalJournal;
+            Config.edit_log_port = originalEditLogPort;
+            Config.query_port = originalQueryPort;
+        }
+    }
+}