This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 17be18138ac branch-4.2: [fix](fe) Refresh backend heartbeat
connections after FQDN IP changes #69450 (#69518)
17be18138ac is described below
commit 17be18138accb13c7d4b91291ffc188b4b633686
Author: bobhan1 <[email protected]>
AuthorDate: Sat Oct 24 10:50:24 2026 +0800
branch-4.2: [fix](fe) Refresh backend heartbeat connections after FQDN IP
changes #69450 (#69518)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #69450
Backport #69450 (commit `4751c5f593cf7f136f2b193621bd5eabde7dc404`) to
`branch-4.2`.
In FQDN mode, a BE Pod IP change can leave FE reusing a heartbeat
connection to the old process, including after an identity error
returned by a successful Thrift RPC. Resolve the destination through the
existing DNSCache, key heartbeat connections by the resolved IP, and
preserve the original backend hostname and ID in the request. Clear the
previous endpoint's idle connections when the address changes, clean up
removed backends, and invalidate unsuccessful heartbeat connections
before retrying. 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 36 targeted FE tests pass on branch-4.2 with 0 failures, errors, or
skips; the repository FE UT script finishes with BUILD SUCCESS. This
includes all 8 new heartbeat handler tests and both live loopback Thrift
connection-switching cases.
```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
```
Both changed Java files pass Checkstyle 9.3 using branch-4.2's
repository configuration. `git diff --check upstream/branch-4.2...HEAD`
passes. The production-code added/removed lines and the new test file
match upstream.
### 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 71e053995c1..1f646622e68 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
@@ -60,6 +60,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;
@@ -80,6 +81,7 @@ public class HeartbeatMgr extends MasterDaemon {
private SystemInfoService nodeMgr;
private HeartbeatFlags heartbeatFlags;
private final ExecutorService abortTxnExecutor;
+ private final Map<Long, TNetworkAddress> backendHeartbeatAddresses =
Maps.newConcurrentMap();
private static volatile AtomicReference<TMasterInfo> masterInfo = new
AtomicReference<>();
@@ -128,6 +130,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));
@@ -270,8 +273,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;
@@ -319,6 +333,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 {
@@ -343,7 +371,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();
@@ -379,6 +406,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(),
@@ -394,6 +422,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]
