From 19535c6f7a98767dae04ec181f8f6371c984fac1 Mon Sep 17 00:00:00 2001 From: shuke <37901441+shuke987@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:10:02 +0800 Subject: [PATCH 1/2] [fix](fe) Keep the mocked FE RPC listener bound through startup ### What problem does this PR solve? Issue Number: None Related PR: None Parallel FE unit tests could lose a probed RPC port before Thrift started, causing FE startup to exit the test JVM. Bind an ephemeral listener once, publish the assigned port before catalog initialization, and pass the live listener to the blocking Thrift server. ### Release note None ### Check List (For Author) - Test: ThriftServerTest and CheckRowPolicyTest passed (12 tests). FE Checkstyle passed. - Behavior changed: No user-facing change. - Does this need documentation: No. --- .../main/java/org/apache/doris/DorisFE.java | 24 +++++-- .../org/apache/doris/common/ThriftServer.java | 57 +++++++--------- .../tls/server/FeServerStarterFactory.java | 6 ++ .../tls/server/FeServerStarterProvider.java | 10 +++ .../server/OssFeServerStarterProvider.java | 18 ++++- .../apache/doris/common/ThriftServerTest.java | 65 +++++++++++++++++++ .../org/apache/doris/utframe/DemoTest.java | 20 ++++++ .../apache/doris/utframe/MockedFrontend.java | 18 ++--- .../doris/utframe/TestWithFeService.java | 6 +- .../apache/doris/utframe/UtFrameUtils.java | 4 +- 10 files changed, 170 insertions(+), 58 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/common/ThriftServerTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java index f341a486ff998f..03615dafb9e369 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java +++ b/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(rpcListener == null); 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,7 +330,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star } } - private static void checkAllPorts() throws IOException { + private static void checkAllPorts(boolean checkRpc) throws IOException { if (Config.enable_check_compatibility_mode) { // The compatibility mode does not need to listen ports. return; @@ -341,7 +352,7 @@ private static void checkAllPorts() throws IOException { "Query port", NetUtils.QUERY_PORT_SUGGESTION)) { throw new IOException("port " + Config.query_port + " already in use"); } - if (!NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.rpc_port, + if (checkRpc && !NetUtils.isPortAvailable(FrontendOptions.getLocalHostAddress(), Config.rpc_port, "Rpc port", NetUtils.RPC_PORT_SUGGESTION)) { throw new IOException("port " + Config.rpc_port + " already in use"); } @@ -717,6 +728,7 @@ private static void setFuzzyForCatalog() { } public static class StartupOptions { + public boolean useEphemeralRpcPort; public boolean enableHttpServer = true; public boolean enableQeService = true; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/ThriftServer.java b/fe/fe-core/src/main/java/org/apache/doris/common/ThriftServer.java index cdc4bba71d3bb9..60ed80c8abdf41 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/ThriftServer.java +++ b/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); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterFactory.java b/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterFactory.java index ed87de8365e979..015c87607cb08f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterFactory.java +++ b/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); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterProvider.java b/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterProvider.java index 8d810b498ab480..317acf79a03d4b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tls/server/FeServerStarterProvider.java +++ b/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); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tls/server/OssFeServerStarterProvider.java b/fe/fe-core/src/main/java/org/apache/doris/tls/server/OssFeServerStarterProvider.java index af8ca85faef2b6..34f4caa7776a86 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tls/server/OssFeServerStarterProvider.java +++ b/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); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/ThriftServerTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/ThriftServerTest.java new file mode 100644 index 00000000000000..05718f19a56ef4 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/common/ThriftServerTest.java @@ -0,0 +1,65 @@ +// 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.common; + +import org.apache.doris.thrift.HeartbeatService; +import org.apache.doris.thrift.TBackendInfo; +import org.apache.doris.thrift.THeartbeatResult; +import org.apache.doris.thrift.TMasterInfo; +import org.apache.doris.thrift.TNetworkAddress; +import org.apache.doris.thrift.TStatus; +import org.apache.doris.thrift.TStatusCode; + +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.transport.TSocket; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.net.BindException; +import java.net.InetAddress; +import java.net.ServerSocket; + +class ThriftServerTest { + @Test + void servesQueuedRpcOnBoundSocket() throws Exception { + InetAddress loopback = InetAddress.getLoopbackAddress(); + try (ServerSocket listener = new ServerSocket(0, 8, loopback); + ServerSocket competitor = new ServerSocket(); + TSocket client = new TSocket(loopback.getHostAddress(), listener.getLocalPort(), 5000)) { + Assertions.assertThrows(BindException.class, () -> competitor.bind(listener.getLocalSocketAddress())); + // Queue the connection before start(). Replacing the listener would discard this connection. + client.open(); + HeartbeatService.Iface heartbeat = master -> new THeartbeatResult(new TStatus(TStatusCode.OK), + new TBackendInfo(master.getNetworkAddress().getPort(), 80)); + ThriftServer server = new ThriftServer(listener.getLocalPort(), + new HeartbeatService.Processor<>(heartbeat)); + server.start(listener); + try { + HeartbeatService.Client service = new HeartbeatService.Client(new TBinaryProtocol(client)); + TMasterInfo master = new TMasterInfo(new TNetworkAddress("localhost", listener.getLocalPort()), 1, 1); + THeartbeatResult reply = service.heartbeat(master); + Assertions.assertEquals(listener.getLocalPort(), reply.getBackendInfo().getBePort()); + } finally { + client.close(); + server.stop(); + server.join(); + } + Assertions.assertTrue(listener.isClosed()); + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoTest.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoTest.java index 2d5fd8638527f6..60726bca747caf 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoTest.java @@ -30,7 +30,13 @@ import org.apache.doris.planner.PlanFragment; import org.apache.doris.planner.Planner; import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.thrift.FrontendService; +import org.apache.doris.thrift.TFrontendPingFrontendRequest; +import org.apache.doris.thrift.TFrontendPingFrontendResult; +import org.apache.doris.thrift.TFrontendPingFrontendStatusCode; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.transport.TSocket; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -52,6 +58,20 @@ protected void runBeforeAll() throws Exception { FeConstants.default_scheduler_interval_millisecond = 10; } + @Test + public void testAdvertisedRpcPort() throws Exception { + Env env = Env.getCurrentEnv(); + try (TSocket socket = new TSocket("127.0.0.1", lastFeRpcPort, 5000)) { + socket.open(); + FrontendService.Client client = new FrontendService.Client(new TBinaryProtocol(socket)); + TFrontendPingFrontendRequest ping = new TFrontendPingFrontendRequest(env.getClusterId(), env.getToken()); + ping.setDeployMode(env.getDeployMode()); + TFrontendPingFrontendResult response = client.ping(ping); + Assertions.assertEquals(TFrontendPingFrontendStatusCode.OK, response.getStatus()); + Assertions.assertEquals(lastFeRpcPort, response.getRpcPort()); + } + } + @Test public void testCreateDbAndTable() throws Exception { // 1. create connect context diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/MockedFrontend.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/MockedFrontend.java index 089c16353d60ca..938f9a28596892 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/utframe/MockedFrontend.java +++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/MockedFrontend.java @@ -19,7 +19,7 @@ import org.apache.doris.DorisFE; import org.apache.doris.DorisFE.StartupOptions; -import org.apache.doris.catalog.Env; +import org.apache.doris.common.Config; import org.apache.doris.common.util.DatasourcePrintableMap; import com.google.common.base.Strings; @@ -77,7 +77,7 @@ public class MockedFrontend { MIN_FE_CONF = Maps.newHashMap(); MIN_FE_CONF.put("sys_log_level", "INFO"); MIN_FE_CONF.put("http_port", "8030"); - MIN_FE_CONF.put("rpc_port", "9020"); + MIN_FE_CONF.put("rpc_port", "0"); MIN_FE_CONF.put("query_port", "9030"); MIN_FE_CONF.put("arrow_flight_sql_port", "8070"); MIN_FE_CONF.put("edit_log_port", "9010"); @@ -177,6 +177,7 @@ public FERunnable(MockedFrontend frontend, String[] args) { @Override public void run() { StartupOptions options = new StartupOptions(); + options.useEphemeralRpcPort = frontend.getRpcPort() == 0; // For FE unit tests, we don't need these 2 servers. // And it also cost time to start up. options.enableHttpServer = false; @@ -192,14 +193,15 @@ public void start(String[] args) throws FeStartException, NotInitException { } Thread feThread = new Thread(new FERunnable(this, args), FE_PROCESS); feThread.start(); - // wait the catalog to be ready until timeout (30 seconds) - waitForCatalogReady(30 * 1000); + // wait for the FE services to be ready until timeout (30 seconds) + waitForServerReady(30 * 1000); + finalFeConf.put("rpc_port", String.valueOf(Config.rpc_port)); } - private void waitForCatalogReady(long timeoutMs) throws FeStartException { + private void waitForServerReady(long timeoutMs) throws FeStartException { long left = timeoutMs; - while (!Env.getCurrentEnv().isReady() && left > 0) { - System.out.println("catalog is not ready"); + while (!DorisFE.isServerReady() && left > 0) { + System.out.println("FE services are not ready"); try { Thread.sleep(100); } catch (InterruptedException e) { @@ -208,7 +210,7 @@ private void waitForCatalogReady(long timeoutMs) throws FeStartException { left -= 100; } - if (left <= 0 && !Env.getCurrentEnv().isReady()) { + if (left <= 0 && !DorisFE.isServerReady()) { throw new FeStartException("fe start failed"); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java index 72a276b631ce5c..2d0afb8c4d58a4 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java +++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java @@ -372,14 +372,12 @@ protected int startFEServerWithoutRetry(String runningDir) System.out.println("CREATE FE SERVER DIR: " + Config.custom_config_dir); int feHttpPort = findValidPort(); - int feRpcPort = findValidPort(); int feQueryPort = findValidPort(); int arrowFlightSqlPort = findValidPort(); int feEditLogPort = findValidPort(); Map feConfMap = Maps.newHashMap(); // set additional fe config feConfMap.put("http_port", String.valueOf(feHttpPort)); - feConfMap.put("rpc_port", String.valueOf(feRpcPort)); feConfMap.put("query_port", String.valueOf(feQueryPort)); feConfMap.put("arrow_flight_sql_port", String.valueOf(arrowFlightSqlPort)); feConfMap.put("edit_log_port", String.valueOf(feEditLogPort)); @@ -388,8 +386,8 @@ protected int startFEServerWithoutRetry(String runningDir) MockedFrontend frontend = new MockedFrontend(); frontend.init(dorisHome + "/" + runningDir, feConfMap); frontend.start(new String[0]); - lastFeRpcPort = feRpcPort; - return feRpcPort; + lastFeRpcPort = frontend.getRpcPort(); + return lastFeRpcPort; } protected void createDorisCluster() diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/UtFrameUtils.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/UtFrameUtils.java index f5bbc21c97c809..dff2441c20eb16 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/utframe/UtFrameUtils.java +++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/UtFrameUtils.java @@ -150,7 +150,6 @@ private static int startFEServerWithoutRetry(String runningDir) throws EnvVarNot } int feHttpPort = findValidPort(); - int feRpcPort = findValidPort(); int feQueryPort = findValidPort(); int arrowFlightSqlPort = findValidPort(); int feEditLogPort = findValidPort(); @@ -160,14 +159,13 @@ private static int startFEServerWithoutRetry(String runningDir) throws EnvVarNot Map feConfMap = Maps.newHashMap(); // set additional fe config feConfMap.put("http_port", String.valueOf(feHttpPort)); - feConfMap.put("rpc_port", String.valueOf(feRpcPort)); feConfMap.put("query_port", String.valueOf(feQueryPort)); feConfMap.put("arrow_flight_sql_port", String.valueOf(arrowFlightSqlPort)); feConfMap.put("edit_log_port", String.valueOf(feEditLogPort)); feConfMap.put("tablet_create_timeout_second", "10"); frontend.init(dorisHome + "/" + runningDir, feConfMap); frontend.start(new String[0]); - return feRpcPort; + return frontend.getRpcPort(); } public static void createDorisCluster(String runningDir) throws InterruptedException, NotInitException, From ac6757ece36d1b919085eb74df8cf8692158b173 Mon Sep 17 00:00:00 2001 From: shuke <37901441+shuke987@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:45:56 +0800 Subject: [PATCH 2/2] [fix](fe) Check startup ports only for active listeners ### Background Mocked FE tests use a local journal and disable the query and HTTP services. Startup still tried binding their configured ports before starting the catalog. ### What problem does this PR solve? Issue Number: None Related PR: None PR Type: Bugfix AnotherDemoTest exited when the edit-log port was occupied. RestoreCommandTest exited at the same startup check for the query port. Use the configured journal and startup options to select listener checks. Enabled services still detect port conflicts before startup. ### Testing Held the edit-log and query sockets open during preflight. The previous implementation failed for a local journal and disabled query service. The new implementation passes and rejects the occupied port when BDB or the query service is enabled. Ran AnotherDemoTest, RestoreCommandTest and the RPC callback test concurrently in separate local JVMs. FE Core Checkstyle passed. ### Release note None ### Check List (For Author) - Behavior changed: Yes. Port preflight follows the selected journal and services. - Does this need documentation: No. - Pre-submission status: Ready. ### AI Disclosure - Tool: OpenAI Codex. - Generated: Code, regression test and validation. --- .../main/java/org/apache/doris/DorisFE.java | 29 +++++---- .../org/apache/doris/DorisFEPortTest.java | 60 +++++++++++++++++++ 2 files changed, 76 insertions(+), 13 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/DorisFEPortTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java index 03615dafb9e369..e7f6e788d50a2c 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java +++ b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java @@ -233,7 +233,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star FrontendOptions.init(); - checkAllPorts(rpcListener == null); + checkAllPorts(options); if (Config.enable_bdbje_debug_mode) { // Start in BDB Debug mode @@ -330,33 +330,36 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star } } - private static void checkAllPorts(boolean checkRpc) 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 (checkRpc && !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"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/DorisFEPortTest.java b/fe/fe-core/src/test/java/org/apache/doris/DorisFEPortTest.java new file mode 100644 index 00000000000000..560ee083cebfa5 --- /dev/null +++ b/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; + } + } +}