This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 72072a30a2b branch-4.1: [fix](fe) Refresh backend heartbeat
connections after FQDN IP changes (#69496)
72072a30a2b is described below
commit 72072a30a2bda5c7f9e055211e34020a081bde42
Author: bobhan1 <[email protected]>
AuthorDate: Mon Oct 26 09:58:42 2026 +0800
branch-4.1: [fix](fe) Refresh backend heartbeat connections after FQDN IP
changes (#69496)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #69450
Problem Summary:
Backport #69450 to `branch-4.1`, cherry-picked with `-x` from
`4751c5f593cf7f136f2b193621bd5eabde7dc404`.
In FQDN mode, BE Pod IP reuse can leave FE reusing a heartbeat socket
connected to the wrong process even after DNS converges. Resolve
heartbeat destinations through FE DNSCache, switch and clean up endpoint
pools when cached addresses change or backends are removed, and
invalidate connections after unsuccessful heartbeat responses. Preserve
the configured backend hostname and ID in the heartbeat request.
Recovery follows the existing DNSCache refresh policy.
The member-declaration conflict is adapted to the 4.1 HeartbeatMgr
layout. Both changed files preserve all additions and deletions from the
upstream fix.
### 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. Heartbeat connections follow cached IP changes and are
discarded after application errors.
- Does this need documentation?
- [x] No.
- [ ] Yes.
Validation:
All 36 targeted FE tests pass on branch-4.1, with 0 failures, errors, or
skips. The FE UT script finishes with BUILD SUCCESS. This includes all 8
BackendHeartbeatHandlerTest cases, including the two live Thrift
connection tests.
```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 with the repository
configuration. `git diff --check upstream/branch-4.1...HEAD` passes.
### 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]
