This is an automated email from the ASF dual-hosted git repository.
deardeng 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 0b0ae091531 [fix](cloud) Bound abort task submissions for non-normal
clusters (#68066)
0b0ae091531 is described below
commit 0b0ae09153153c73d22e1b2e225f3ae6061a3d2c
Author: deardeng <[email protected]>
AuthorDate: Fri Sep 18 16:14:03 2026 +0800
[fix](cloud) Bound abort task submissions for non-normal clusters (#68066)
Problem Summary: When a BE in a non-normal cloud cluster stays offline,
every failed heartbeat after the lost-heartbeat threshold submits
another transaction cleanup task. Each task queries
get_prepare_txn_by_coordinator, which scans the instance transaction
metadata even after cleanup finishes. Restrict these submissions to
lost-heartbeat durations in [T, 2T), where T is
abort_txn_after_lost_heartbeat_time_second. Keep the existing behavior
for NORMAL cloud clusters and non-cloud deployments. This deliberately
stops further submissions after the window even if earlier cleanup
attempts failed.
### Release note
Limit repeated transaction cleanup task submissions for offline BEs in
non-normal cloud clusters to a bounded window (5 to 10 minutes after the
last successful heartbeat with the default configuration).
### Check List (For Author)
- Test: Add HeartbeatMgrTest coverage for all four non-normal cluster
states, durations before/inside/after the window, NORMAL and non-cloud
behavior, replay, and missing successful heartbeats. Tests and local
compilation were not run as requested. Checkstyle on both changed files
and git diff --check passed.
- Behavior changed: Yes, stop submitting BE-down cleanup tasks for
non-normal cloud clusters once the lost-heartbeat duration reaches twice
the threshold.
- Does this need documentation: No
---
.../java/org/apache/doris/system/HeartbeatMgr.java | 11 ++--
.../org/apache/doris/system/HeartbeatMgrTest.java | 60 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 3 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 dbd36715d53..51347942f25 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
@@ -19,6 +19,7 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FsBroker;
+import org.apache.doris.cloud.proto.Cloud.ClusterStatus;
import org.apache.doris.common.ClientPool;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
@@ -217,9 +218,13 @@ public class HeartbeatMgr extends MasterDaemon {
} else {
// invalid all connections cached in ClientPool
ClientPool.backendPool.clearPool(new
TNetworkAddress(be.getHost(), be.getBePort()));
- if (!isReplay && System.currentTimeMillis() -
be.getLastUpdateMs()
- >=
Config.abort_txn_after_lost_heartbeat_time_second * 1000L
- && be.getLastUpdateMs() > 0) {
+ long lostTimeMs = System.currentTimeMillis() -
be.getLastUpdateMs();
+ long timeoutMs =
Config.abort_txn_after_lost_heartbeat_time_second * 1000L;
+ boolean nonNormalCloudCluster = Config.isCloudMode()
+ &&
!ClusterStatus.NORMAL.name().equals(be.getCloudClusterStatus());
+ // Bound repeated MS scans for inactive clusters to
[timeoutMs, 2 * timeoutMs).
+ if (!isReplay && be.getLastUpdateMs() > 0 &&
lostTimeMs >= timeoutMs
+ && (!nonNormalCloudCluster || lostTimeMs < 2 *
timeoutMs)) {
submitAbortTxnTaskByExecutor(() ->
Env.getCurrentGlobalTransactionMgr()
.abortTxnWhenCoordinateBeDown(be.getId(),
be.getHost(), 100), "down");
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/system/HeartbeatMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/system/HeartbeatMgrTest.java
index 91b21bca6f0..4dd8966eacd 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/system/HeartbeatMgrTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/system/HeartbeatMgrTest.java
@@ -19,9 +19,11 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FsBroker;
+import org.apache.doris.cloud.proto.Cloud.ClusterStatus;
import org.apache.doris.common.ClientPool;
import org.apache.doris.common.Config;
import org.apache.doris.common.GenericPool;
+import org.apache.doris.common.ThreadPoolManager;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.system.HeartbeatMgr.BrokerHeartbeatHandler;
import org.apache.doris.system.HeartbeatMgr.FrontendHeartbeatHandler;
@@ -46,6 +48,8 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReference;
public class HeartbeatMgrTest {
@@ -188,4 +192,60 @@ public class HeartbeatMgrTest {
}
}
+ @Test
+ public void testNonNormalCloudClusterAbortWindow() throws Exception {
+ SystemInfoService nodeMgr = Mockito.mock(SystemInfoService.class);
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(nodeMgr.getBackend(1L)).thenReturn(backend);
+ Mockito.when(backend.getHost()).thenReturn("127.0.0.1");
+ ThreadPoolExecutor executor = Mockito.mock(ThreadPoolExecutor.class);
+ BackendHbResponse response = new BackendHbResponse(1L, "127.0.0.1", 0,
"heartbeat failed");
+ Method handleHbResponse = HeartbeatMgr.class.getDeclaredMethod(
+ "handleHbResponse", HeartbeatResponse.class, boolean.class);
+ handleHbResponse.setAccessible(true);
+ long timeoutMs = Config.abort_txn_after_lost_heartbeat_time_second *
1000L;
+
+ try (MockedStatic<Config> config = Mockito.mockStatic(Config.class);
+ MockedStatic<ThreadPoolManager> threadPools =
Mockito.mockStatic(ThreadPoolManager.class)) {
+ config.when(Config::isCloudMode).thenReturn(true);
+ threadPools.when(() -> ThreadPoolManager.newDaemonFixedThreadPool(
+ Mockito.anyInt(), Mockito.anyInt(), Mockito.anyString(),
Mockito.anyBoolean()))
+ .thenReturn(executor);
+ HeartbeatMgr mgr = new HeartbeatMgr(nodeMgr, false);
+ for (ClusterStatus status : new ClusterStatus[]
{ClusterStatus.UNKNOWN, ClusterStatus.SUSPENDED,
+ ClusterStatus.TO_RESUME, ClusterStatus.MANUAL_SHUTDOWN}) {
+ Mockito.clearInvocations(executor);
+
Mockito.when(backend.getCloudClusterStatus()).thenReturn(status.name());
+
+
Mockito.when(backend.getLastUpdateMs()).thenReturn(System.currentTimeMillis() -
timeoutMs / 2);
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.never()).submit(Mockito.any(Runnable.class));
+
+
Mockito.when(backend.getLastUpdateMs()).thenReturn(System.currentTimeMillis() -
timeoutMs * 3 / 2);
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.times(1)).submit(Mockito.any(Runnable.class));
+ handleHbResponse.invoke(mgr, response, true);
+ Mockito.verify(executor,
Mockito.times(1)).submit(Mockito.any(Runnable.class));
+
+
Mockito.when(backend.getLastUpdateMs()).thenReturn(System.currentTimeMillis() -
timeoutMs * 3);
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.times(1)).submit(Mockito.any(Runnable.class));
+ }
+
+ Mockito.clearInvocations(executor);
+
Mockito.when(backend.getCloudClusterStatus()).thenReturn(ClusterStatus.NORMAL.name());
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.times(1)).submit(Mockito.any(Runnable.class));
+
+ config.when(Config::isCloudMode).thenReturn(false);
+
Mockito.when(backend.getCloudClusterStatus()).thenReturn(ClusterStatus.SUSPENDED.name());
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.times(2)).submit(Mockito.any(Runnable.class));
+
+ Mockito.when(backend.getLastUpdateMs()).thenReturn(0L);
+ handleHbResponse.invoke(mgr, response, false);
+ Mockito.verify(executor,
Mockito.times(2)).submit(Mockito.any(Runnable.class));
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
