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 4bea404e865 [fix](cloud) Bound abort task submissions for non-normal
clusters (#68066) (#68947)
4bea404e865 is described below
commit 4bea404e86535d96c5a10977bafe62822f559564
Author: deardeng <[email protected]>
AuthorDate: Mon Oct 12 19:18:15 2026 +0800
[fix](cloud) Bound abort task submissions for non-normal clusters (#68066)
(#68947)
pick from https://github.com/apache/doris/pull/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.
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).
- 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
(cherry picked from commit 0b0ae09153153c73d22e1b2e225f3ae6061a3d2c)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../java/org/apache/doris/system/HeartbeatMgr.java | 11 +++-
.../org/apache/doris/system/HeartbeatMgrTest.java | 64 ++++++++++++++++++++++
2 files changed, 72 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 5e13e9191af..71e053995c1 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;
@@ -214,9 +215,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 0ad1b0bb645..0ce265c083b 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,7 +19,10 @@ 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.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;
@@ -43,6 +46,11 @@ import org.apache.thrift.TException;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.ThreadPoolExecutor;
public class HeartbeatMgrTest {
@@ -181,4 +189,60 @@ public class HeartbeatMgrTest {
Assert.assertEquals(HbStatus.OK, hbResponse.getStatus());
}
+ @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]
