This is an automated email from the ASF dual-hosted git repository.
bobhan1 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 4751c5f593c [fix](fe) Refresh backend heartbeat connections after FQDN
IP changes (#69450)
4751c5f593c is described below
commit 4751c5f593cf7f136f2b193621bd5eabde7dc404
Author: bobhan1 <[email protected]>
AuthorDate: Fri Oct 23 10:30:12 2026 +0800
[fix](fe) Refresh backend heartbeat connections after FQDN IP changes
(#69450)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
In FQDN mode, a BE Pod restart can leave FE with a stale hostname-to-IP
mapping. If another BE takes the old IP, a new heartbeat connection can
reach that process. The process can return an identity error through a
successful Thrift RPC, and FE returns the socket to the pool. Later
heartbeats reuse it even after DNS has converged. A wrong process can
also accept the heartbeat while its own DNS view is stale.
Resolve heartbeat destinations through the existing FE DNSCache and
borrow connections using the resolved IP, while preserving the original
backend hostname and ID in TMasterInfo. Close the previous endpoint's
idle pool when the address changes, and clean up pools for removed
backends. Invalidate connections on unsuccessful heartbeat responses so
the existing retry can reconnect. Recovery follows the existing DNSCache
refresh policy.
### Release note
Fix persistent backend heartbeat failures after BE Pod IP changes in
FQDN mode.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test
- [ ] No need to test or manual test.
- Behavior changed:
- [ ] No.
- [x] Yes. Heartbeats switch connections when the cached IP changes and
discard connections after application errors.
- Does this need documentation?
- [x] No.
- [ ] Yes.
Validation:
All 39 targeted FE tests pass with 0 failures, errors, or skips. Both
changed Java files pass Checkstyle 9.3 using the repository's Checkstyle
configuration. The FE UT script finishes with BUILD SUCCESS.
```bash
./run-fe-ut.sh --run
org.apache.doris.system.BackendHeartbeatHandlerTest,org.apache.doris.system.HeartbeatMgrTest,org.apache.doris.common.GenericPoolTest,org.apache.doris.common.DNSCacheTest,org.apache.doris.rpc.BackendServiceProxyTest
```
The new tests cover stable-address reuse, preserved FQDN identity,
rejected replies, RPC exceptions, DNS lookup failure, IP mode,
removed-backend cleanup, and address changes. Two tests use live
loopback Thrift servers and wait for the old connection to close,
including a healthy old connection that continues accepting heartbeats.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../java/org/apache/doris/system/HeartbeatMgr.java | 34 ++-
.../doris/system/BackendHeartbeatHandlerTest.java | 309 +++++++++++++++++++++
2 files changed, 341 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
index 736af4b558e..345cd287be8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
@@ -61,6 +61,7 @@ import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -84,6 +85,7 @@ public class HeartbeatMgr extends MasterDaemon {
private final ExecutorService abortTxnExecutor;
private final ExecutorService changelogTtlExecutor;
private final AtomicBoolean changelogTtlRefreshPending = new
AtomicBoolean();
+ private final Map<Long, TNetworkAddress> backendHeartbeatAddresses =
Maps.newConcurrentMap();
private long lastChangelogTtlRefreshMs;
@@ -170,6 +172,7 @@ public class HeartbeatMgr extends MasterDaemon {
LOG.warn("can not get backends", e);
return;
}
+ clearRemovedBackendHeartbeatPools();
for (Backend backend : bes) {
BackendHeartbeatHandler handler = new
BackendHeartbeatHandler(backend, feInfos);
hbResponses.add(executor.submit(handler));
@@ -312,8 +315,19 @@ public class HeartbeatMgr extends MasterDaemon {
}
}
+ void clearRemovedBackendHeartbeatPools() {
+ backendHeartbeatAddresses.entrySet().removeIf(entry -> {
+ // Removed BEs will never borrow again, so their idle connections
must be closed here.
+ if (nodeMgr.getBackend(entry.getKey()) == null) {
+ ClientPool.backendHeartbeatPool.clearPool(entry.getValue());
+ return true;
+ }
+ return false;
+ });
+ }
+
// backend heartbeat
- private class BackendHeartbeatHandler implements
Callable<HeartbeatResponse> {
+ class BackendHeartbeatHandler implements Callable<HeartbeatResponse> {
private Backend backend;
private List<TFrontendInfo> feInfos;
@@ -361,6 +375,20 @@ public class HeartbeatMgr extends MasterDaemon {
}
THeartbeatResult result;
if (!FeConstants.runningUnitTest) {
+ if (Config.enable_fqdn_mode) {
+ String realIp =
Env.getCurrentEnv().getDnsCache().get(backend.getHost());
+ // DNSCache returns an empty address when the first
lookup fails.
+ if (realIp.isEmpty()) {
+ throw new UnknownHostException("Failed to resolve
hostname: " + backend.getHost());
+ }
+ beAddr = new TNetworkAddress(realIp,
backend.getHeartbeatPort());
+ }
+ TNetworkAddress previousAddress =
backendHeartbeatAddresses.put(backendId, beAddr);
+ // A reused Pod IP can lead to a healthy connection to the
wrong BE, even when
+ // its heartbeat succeeds. Switch pools as soon as
DNSCache observes the new IP.
+ if (previousAddress != null &&
!previousAddress.equals(beAddr)) {
+
ClientPool.backendHeartbeatPool.clearPool(previousAddress);
+ }
client =
ClientPool.backendHeartbeatPool.borrowObject(beAddr);
result = client.heartbeat(copiedMasterInfo);
} else {
@@ -385,7 +413,6 @@ public class HeartbeatMgr extends MasterDaemon {
result.getStatus().addToErrorMsgs("debug point
HeartbeatMgr.deadBeIds set dead be");
}
- ok = true;
if (result.getStatus().getStatusCode() == TStatusCode.OK) {
TBackendInfo tBackendInfo = result.getBackendInfo();
int bePort = tBackendInfo.getBePort();
@@ -421,6 +448,7 @@ public class HeartbeatMgr extends MasterDaemon {
fragmentNum, lastFragmentUpdateTime, isShutDown,
arrowFlightSqlPort, beMemory);
response.setPaimonRustReaderSupported(tBackendInfo.isSetSupportsPaimonRustReader()
&& tBackendInfo.isSupportsPaimonRustReader());
+ ok = true;
return response;
} else {
return new BackendHbResponse(backendId, backend.getHost(),
backend.getLastUpdateMs(),
@@ -436,6 +464,8 @@ public class HeartbeatMgr extends MasterDaemon {
if (ok) {
ClientPool.backendHeartbeatPool.returnObject(beAddr,
client);
} else {
+ // Identity errors are normal RPC replies. Reusing
that socket would keep
+ // contacting the wrong BE after DNS has converged.
ClientPool.backendHeartbeatPool.invalidateObject(beAddr, client);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/system/BackendHeartbeatHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/system/BackendHeartbeatHandlerTest.java
new file mode 100644
index 00000000000..81592c7c213
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/system/BackendHeartbeatHandlerTest.java
@@ -0,0 +1,309 @@
+// 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.system;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TokenManager;
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DNSCache;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.GenericPool;
+import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.system.HeartbeatResponse.HbStatus;
+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.commons.pool2.impl.GenericKeyedObjectPoolConfig;
+import org.apache.thrift.TException;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TServerSocket;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+class BackendHeartbeatHandlerTest {
+ private static final String HOST = "be-0.be-service.test";
+ private static final String OLD_IP = "127.0.0.1";
+ private static final String NEW_IP = "127.0.0.2";
+
+ private final DNSCache dnsCache = Mockito.mock(DNSCache.class);
+ private final SystemInfoService nodeMgr =
Mockito.mock(SystemInfoService.class);
+ @SuppressWarnings("unchecked")
+ private final GenericPool<HeartbeatService.Client> pool =
Mockito.mock(GenericPool.class);
+ private final HeartbeatService.Client client =
Mockito.mock(HeartbeatService.Client.class);
+ private GenericPool<HeartbeatService.Client> originalPool;
+ private boolean originalFqdnMode;
+ private boolean originalRunningUnitTest;
+ private AtomicReference<TMasterInfo> masterInfo;
+ private TMasterInfo originalMasterInfo;
+ private MockedStatic<Env> mockedEnv;
+ private MockedStatic<FrontendOptions> mockedFrontendOptions;
+ private HeartbeatMgr mgr;
+ private Backend backend;
+
+ @BeforeEach
+ @SuppressWarnings("unchecked")
+ void setUp() throws Exception {
+ originalPool = ClientPool.backendHeartbeatPool;
+ originalFqdnMode = Config.enable_fqdn_mode;
+ originalRunningUnitTest = FeConstants.runningUnitTest;
+ ClientPool.backendHeartbeatPool = pool;
+ Config.enable_fqdn_mode = true;
+ FeConstants.runningUnitTest = false;
+
+ Field field = HeartbeatMgr.class.getDeclaredField("masterInfo");
+ field.setAccessible(true);
+ masterInfo = (AtomicReference<TMasterInfo>) field.get(null);
+ originalMasterInfo = masterInfo.get();
+
+ Env env = Mockito.mock(Env.class);
+ TokenManager tokenManager = Mockito.mock(TokenManager.class);
+ Mockito.when(tokenManager.acquireToken()).thenReturn("auth-token");
+ Mockito.when(env.getTokenManager()).thenReturn(tokenManager);
+ Mockito.when(env.getDnsCache()).thenReturn(dnsCache);
+ mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ mockedFrontendOptions = Mockito.mockStatic(FrontendOptions.class);
+
mockedFrontendOptions.when(FrontendOptions::getLocalHostAddress).thenReturn(OLD_IP);
+
+ Mockito.when(dnsCache.get(HOST)).thenReturn(OLD_IP);
+
Mockito.when(pool.borrowObject(Mockito.any(TNetworkAddress.class))).thenReturn(client);
+
Mockito.when(client.heartbeat(Mockito.any(TMasterInfo.class))).thenReturn(success(123L));
+ backend = new Backend(1L, HOST, 9050);
+ Mockito.when(nodeMgr.getBackend(backend.getId())).thenReturn(backend);
+ mgr = new HeartbeatMgr(nodeMgr, false);
+ mgr.setMaster(1, "token", 1L);
+ }
+
+ @AfterEach
+ void tearDown() {
+ ClientPool.backendHeartbeatPool = originalPool;
+ Config.enable_fqdn_mode = originalFqdnMode;
+ FeConstants.runningUnitTest = originalRunningUnitTest;
+ masterInfo.set(originalMasterInfo);
+ mockedFrontendOptions.close();
+ mockedEnv.close();
+ }
+
+ private BackendHbResponse heartbeat() {
+ return (BackendHbResponse) mgr.new BackendHeartbeatHandler(backend,
Collections.emptyList()).call();
+ }
+
+ private static THeartbeatResult success(long startTime) {
+ TBackendInfo info = new TBackendInfo();
+ info.setBePort(9060);
+ info.setHttpPort(8040);
+ info.setBeStartTime(startTime);
+ return new THeartbeatResult(new TStatus(TStatusCode.OK), info);
+ }
+
+ @Test
+ void testStableAddressReusesPoolAndPreservesFqdnIdentity() throws
Exception {
+ Assertions.assertEquals(HbStatus.OK, heartbeat().getStatus());
+ Assertions.assertEquals(HbStatus.OK, heartbeat().getStatus());
+ TNetworkAddress address = new TNetworkAddress(OLD_IP, 9050);
+ Mockito.verify(pool, Mockito.times(2)).borrowObject(address);
+ Mockito.verify(pool, Mockito.times(2)).returnObject(address, client);
+ Mockito.verify(pool, Mockito.never()).clearPool(Mockito.any());
+ ArgumentCaptor<TMasterInfo> requests =
ArgumentCaptor.forClass(TMasterInfo.class);
+ Mockito.verify(client, Mockito.times(2)).heartbeat(requests.capture());
+ for (TMasterInfo request : requests.getAllValues()) {
+ Assertions.assertEquals(HOST, request.getBackendIp());
+ Assertions.assertEquals(backend.getId(), request.getBackendId());
+ }
+ }
+
+ @Test
+ void testRejectedHeartbeatInvalidatesConnectionOnBothAttempts() throws
Exception {
+ TStatus status = new TStatus(TStatusCode.INTERNAL_ERROR);
+ status.addToErrorMsgs("the host recorded in master does not map to
this BE");
+ Mockito.when(client.heartbeat(Mockito.any())).thenReturn(new
THeartbeatResult(status, new TBackendInfo()));
+ Assertions.assertEquals(HbStatus.BAD, heartbeat().getStatus());
+ Mockito.verify(pool, Mockito.times(2)).invalidateObject(new
TNetworkAddress(OLD_IP, 9050), client);
+ Mockito.verify(pool, Mockito.never()).returnObject(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ void testRpcExceptionInvalidatesConnectionBeforeRetry() throws Exception {
+ Mockito.when(client.heartbeat(Mockito.any())).thenThrow(new
TException("connection reset"))
+ .thenReturn(success(123L));
+ Assertions.assertEquals(HbStatus.OK, heartbeat().getStatus());
+ TNetworkAddress address = new TNetworkAddress(OLD_IP, 9050);
+ Mockito.verify(pool).invalidateObject(address, client);
+ Mockito.verify(pool).returnObject(address, client);
+ }
+
+ @Test
+ void testFailedDnsLookupDoesNotBorrowConnection() {
+ Mockito.when(dnsCache.get(HOST)).thenReturn("");
+ BackendHbResponse response = heartbeat();
+ Assertions.assertEquals(HbStatus.BAD, response.getStatus());
+ Assertions.assertTrue(response.getMsg().contains("Failed to resolve
hostname"));
+ Mockito.verifyNoInteractions(pool);
+ }
+
+ @Test
+ void testIpModeDoesNotUseDnsCache() throws Exception {
+ Config.enable_fqdn_mode = false;
+ backend = new Backend(1L, OLD_IP, 9050);
+ Assertions.assertEquals(HbStatus.OK, heartbeat().getStatus());
+ Mockito.verify(pool).borrowObject(new TNetworkAddress(OLD_IP, 9050));
+ Mockito.verifyNoInteractions(dnsCache);
+ }
+
+ @Test
+ void testRemovedBackendClosesItsIdlePool() {
+ Assertions.assertEquals(HbStatus.OK, heartbeat().getStatus());
+ Mockito.when(nodeMgr.getBackend(backend.getId())).thenReturn(null);
+ mgr.clearRemovedBackendHeartbeatPools();
+ mgr.clearRemovedBackendHeartbeatPools();
+ Mockito.verify(pool).clearPool(new TNetworkAddress(OLD_IP, 9050));
+ }
+
+ @Test
+ void testDnsChangeClosesHealthyConnectionToOldProcess() throws Exception {
+ // Both processes keep their sockets healthy and accept the heartbeat.
Only the DNS
+ // change can move the next heartbeat away from the old process.
+ try (HeartbeatEndpoint oldEndpoint = new HeartbeatEndpoint(OLD_IP, 0,
request -> success(111L));
+ HeartbeatEndpoint newEndpoint = new HeartbeatEndpoint(
+ NEW_IP, oldEndpoint.port(), request -> success(222L)))
{
+ useRealPool(oldEndpoint.port());
+ try {
+ Assertions.assertEquals(111L, heartbeat().getBeStartTime());
+ Assertions.assertEquals(111L, heartbeat().getBeStartTime());
+ Mockito.when(dnsCache.get(HOST)).thenReturn(NEW_IP);
+ Assertions.assertEquals(222L, heartbeat().getBeStartTime());
+ Assertions.assertTrue(oldEndpoint.disconnected.await(5,
TimeUnit.SECONDS));
+ Assertions.assertEquals(2, oldEndpoint.calls.get());
+ Assertions.assertEquals(1, newEndpoint.calls.get());
+ } finally {
+ clearRealPool(oldEndpoint.port());
+ }
+ }
+ }
+
+ @Test
+ void testIdentityMismatchReconnectsToNewProcessWithinRetry() throws
Exception {
+ AtomicReference<String> resolvedIp = new AtomicReference<>(OLD_IP);
+ Mockito.when(dnsCache.get(HOST)).thenAnswer(invocation ->
resolvedIp.get());
+ try (HeartbeatEndpoint wrongEndpoint = new HeartbeatEndpoint(OLD_IP,
0, request -> {
+ // Simulate DNS converging after a stale lookup reached the wrong
BE. The wrong
+ // process stays alive, so the client must close the rejected
socket itself.
+ resolvedIp.set(NEW_IP);
+ TStatus status = new TStatus(TStatusCode.INTERNAL_ERROR);
+ status.addToErrorMsgs("the host recorded in master does not map to
this BE");
+ return new THeartbeatResult(status, new TBackendInfo());
+ }); HeartbeatEndpoint rightEndpoint = new HeartbeatEndpoint(
+ NEW_IP, wrongEndpoint.port(), request -> success(222L))) {
+ useRealPool(wrongEndpoint.port());
+ try {
+ BackendHbResponse response = heartbeat();
+ Assertions.assertEquals(HbStatus.OK, response.getStatus());
+ Assertions.assertEquals(222L, response.getBeStartTime());
+ Assertions.assertTrue(wrongEndpoint.disconnected.await(5,
TimeUnit.SECONDS));
+ Assertions.assertEquals(1, wrongEndpoint.calls.get());
+ Assertions.assertEquals(1, rightEndpoint.calls.get());
+ } finally {
+ clearRealPool(wrongEndpoint.port());
+ }
+ }
+ }
+
+ private void useRealPool(int port) {
+ ClientPool.backendHeartbeatPool = new GenericPool<>(
+ "HeartbeatService", new GenericKeyedObjectPoolConfig(), 2000);
+ backend = new Backend(1L, HOST, port);
+ }
+
+ private void clearRealPool(int port) {
+ ClientPool.backendHeartbeatPool.clearPool(new TNetworkAddress(OLD_IP,
port));
+ ClientPool.backendHeartbeatPool.clearPool(new TNetworkAddress(NEW_IP,
port));
+ }
+
+ private static class HeartbeatEndpoint implements AutoCloseable {
+ private final TServerSocket socket;
+ private final TThreadPoolServer server;
+ private final ExecutorService serving =
Executors.newSingleThreadExecutor();
+ private final ExecutorService workers =
Executors.newSingleThreadExecutor();
+ private final AtomicInteger calls = new AtomicInteger();
+ private final CountDownLatch disconnected = new CountDownLatch(1);
+
+ HeartbeatEndpoint(String ip, int port, HeartbeatService.Iface handler)
throws Exception {
+ socket = new TServerSocket(new ServerSocket(port, 50,
InetAddress.getByName(ip)));
+ HeartbeatService.Iface checkedHandler = request -> {
+ Assertions.assertEquals(HOST, request.getBackendIp());
+ calls.incrementAndGet();
+ return handler.heartbeat(request);
+ };
+ server = new TThreadPoolServer(new TThreadPoolServer.Args(socket)
+ .processor(new
HeartbeatService.Processor<>(checkedHandler)).executorService(workers));
+ CountDownLatch ready = new CountDownLatch(1);
+ TServerEventHandler events =
Mockito.mock(TServerEventHandler.class);
+ Mockito.when(events.createContext(Mockito.any(), Mockito.any()))
+ .thenAnswer(invocation ->
Mockito.mock(ServerContext.class));
+ Mockito.doAnswer(invocation -> {
+ ready.countDown();
+ return null;
+ }).when(events).preServe();
+ Mockito.doAnswer(invocation -> {
+ disconnected.countDown();
+ return null;
+ }).when(events).deleteContext(Mockito.any(), Mockito.any(),
Mockito.any());
+ server.setServerEventHandler(events);
+ serving.submit(server::serve);
+ Assertions.assertTrue(ready.await(5, TimeUnit.SECONDS));
+ }
+
+ int port() {
+ return socket.getServerSocket().getLocalPort();
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ server.stop();
+ socket.close();
+ workers.shutdownNow();
+ serving.shutdownNow();
+ Assertions.assertTrue(workers.awaitTermination(5,
TimeUnit.SECONDS));
+ Assertions.assertTrue(serving.awaitTermination(5,
TimeUnit.SECONDS));
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
