This is an automated email from the ASF dual-hosted git repository.
hello-stephen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ee73acc6131 [fix](fe) Prevent port conflicts during mocked FE startup
(#70526)
ee73acc6131 is described below
commit ee73acc6131e6c6c752eb59e252404590be6ad77
Author: shuke <[email protected]>
AuthorDate: Fri Nov 13 16:08:34 2026 +0800
[fix](fe) Prevent port conflicts during mocked FE startup (#70526)
### Background
FE unit tests start a mocked frontend in each test JVM. It uses a local
file journal and disables the query and HTTP services. Mock backends
still need to call the FE's advertised Thrift RPC port.
### What problem does this PR solve?
Issue Number: None
Related PR: None
PR Type: Bugfix
Mocked FE startup could exit the entire test JVM when another process
occupied a port between selection and startup. The FE now keeps the
ephemeral RPC listener bound until Thrift takes ownership. Startup
checks the other ports only for the selected journal and enabled
services.
### Details
- In one parallel run, Thrift could not bind the previously selected RPC
port. Other runs failed earlier in `DorisFE.checkAllPorts`:
`AnotherDemoTest` reported an occupied edit-log port and
`RestoreCommandTest` an occupied query port. Both forks exited with code
255 before writing their JUnit XML.
- The mocked FE binds an ephemeral RPC listener before catalog
initialization and advertises its assigned port to the backends. The
blocking Thrift server accepts that socket. Providers that need their
own transport retain the port-based startup path.
- Port preflight checks the replication port for a BDB journal, the
query and Arrow Flight SQL ports for an enabled query service, and
HTTP(S) ports for an enabled HTTP server. A local file journal opens no
replication listener.
### Testing
- Held the edit-log and query ports open throughout a startup preflight
test. The previous implementation failed with the local journal and
query service disabled. The updated preflight passes in that mode and
rejects the occupied port when BDB or the query service is enabled.
- Ran `AnotherDemoTest`, `RestoreCommandTest`, and
`DemoTest.testAdvertisedRpcPort` concurrently in three local JVMs. All
four test methods passed. The RPC test connects to the port advertised
to the mock backend.
- A separate three-fork Maven run reproduced a shared plugin-directory
creation failure in `RestoreCommandTest` (`PluginMgr.init`, before the
test wrote its XML). Its mocked FE log identified a filesystem error.
The other four selected classes completed.
- The repository’s `run-fe-ut.sh` selected five test classes with one
fork. All six selected methods passed with zero skips, failures or
errors.
- The earlier RPC verification passed all 12 tests in `ThriftServerTest`
and `CheckRowPolicyTest`. FE Core Checkstyle passed for the updated code
and test.
---
.../src/main/java/org/apache/doris/DorisFE.java | 47 ++++++++++------
.../java/org/apache/doris/common/ThriftServer.java | 57 ++++++++-----------
.../doris/tls/server/FeServerStarterFactory.java | 6 ++
.../doris/tls/server/FeServerStarterProvider.java | 10 ++++
.../tls/server/OssFeServerStarterProvider.java | 18 +++++-
.../java/org/apache/doris/DorisFEPortTest.java | 60 ++++++++++++++++++++
.../org/apache/doris/common/ThriftServerTest.java | 65 ++++++++++++++++++++++
.../java/org/apache/doris/utframe/DemoTest.java | 20 +++++++
.../org/apache/doris/utframe/MockedFrontend.java | 18 +++---
.../apache/doris/utframe/TestWithFeService.java | 6 +-
.../org/apache/doris/utframe/UtFrameUtils.java | 4 +-
11 files changed, 243 insertions(+), 68 deletions(-)
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 f341a486ff9..e7f6e788d50 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.File;
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 class DorisFE {
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 class DorisFE {
FrontendOptions.init();
- // check all port
- checkAllPorts();
+ checkAllPorts(options);
if (Config.enable_bdbje_debug_mode) {
// Start in BDB Debug mode
@@ -256,6 +257,14 @@ public class DorisFE {
}
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 class DorisFE {
// 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 class DorisFE {
}
}
- 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 @@ public class DorisFE {
}
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 cdc4bba71d3..60ed80c8abd 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 class ThriftServer {
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 @@ public class ThriftServer {
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 @@ public class ThriftServer {
}
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 ed87de8365e..015c87607cb 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.doris.qe.ConnectScheduler;
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 final class FeServerStarterFactory {
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 8d810b498ab..317acf79a03 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 @@ package org.apache.doris.tls.server;
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 af8ca85faef..34f4caa7776 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.LogManager;
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 class OssFeServerStarterProvider implements
FeServerStarterProvider {
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 class OssFeServerStarterProvider implements
FeServerStarterProvider {
});
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 00000000000..560ee083ceb
--- /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 00000000000..05718f19a56
--- /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 2d5fd863852..60726bca747 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.OlapScanNode;
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 @@ public class DemoTest extends TestWithFeService {
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 089c16353d6..938f9a28596 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 @@ package org.apache.doris.utframe;
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 class MockedFrontend {
@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 class MockedFrontend {
}
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 @@ public class MockedFrontend {
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 72a276b631c..2d0afb8c4d5 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 @@ public abstract class TestWithFeService {
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<String, String> 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 @@ public abstract class TestWithFeService {
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 f5bbc21c97c..dff2441c20e 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 @@ public class UtFrameUtils {
}
int feHttpPort = findValidPort();
- int feRpcPort = findValidPort();
int feQueryPort = findValidPort();
int arrowFlightSqlPort = findValidPort();
int feEditLogPort = findValidPort();
@@ -160,14 +159,13 @@ public class UtFrameUtils {
Map<String, String> 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,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
