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
commit e0f0406527e17c945ed4c91c0bb60638e45eb335 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]
