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..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 @@ -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; } 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/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; + } + } +} 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,