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 97aa9f48af2 branch-4.1: [fix](cloud) Release warm-up destination on
initialization failure #67924 (#68107)
97aa9f48af2 is described below
commit 97aa9f48af24e903d81561b86466be213a10d402
Author: bobhan1 <[email protected]>
AuthorDate: Sat Sep 19 15:36:58 2026 +0800
branch-4.1: [fix](cloud) Release warm-up destination on initialization
failure #67924 (#68107)
### What problem does this PR solve?
Backport #67924 to branch-4.1 (source commit
f0affa953dff3d0e761bd6bb2f62b231b31a9e26).
When tablet-batch initialization fails, release the warm-up destination
registration and persist the error. ONCE jobs become CANCELLED; PERIODIC
jobs remain PENDING and retry at their existing interval. No BE cleanup
RPC is sent before a BE job has started.
### Conflict resolution
Production code applied unchanged. The only conflict was in
CloudWarmUpJobTest: master migrated this class from JUnit 4 to JUnit 5
in #67396 (050442dfc4757b5f042c31e64efd12ed9e9839d2), while branch-4.1
retains JUnit 4. Keep the target framework and express the two
EnumSource tests as four independent JUnit 4 tests, preserving
ONCE/PERIODIC failure and success coverage.
### Validation
- Passed `./run-fe-ut.sh --run
'org.apache.doris.cloud.CloudWarmUpJobTest,org.apache.doris.cloud.CacheHotspotManagerSchedulerTest,org.apache.doris.cloud.cache.CacheHotspotManagerTest'`:
30 tests, zero failures/errors/skips; BUILD SUCCESS.
- Passed `git diff --check upstream/branch-4.1...HEAD`.
- No live cloud-cluster regression or separate full build was run.
### Release note
Release the destination compute group when cloud warm-up initialization
fails, allowing subsequent warm-up jobs to proceed while preserving
periodic retry scheduling.
---
.../org/apache/doris/cloud/CloudWarmUpJob.java | 42 ++++----
.../org/apache/doris/cloud/CloudWarmUpJobTest.java | 110 +++++++++++++++++++++
2 files changed, 135 insertions(+), 17 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
index e2f58cc49b0..6fbe6188d5e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
@@ -932,23 +932,31 @@ public class CloudWarmUpJob implements Writable {
return;
}
- // Todo: nothing to prepare yet
- this.setJobDone = false;
- this.lastBatchId = -1;
- this.startTimeMs = System.currentTimeMillis();
- // reset clients to ensure we have the latest BE info
- this.beToThriftAddress = null;
- this.beToClient = null;
- this.beToAddr = null;
-
MetricRepo.updateClusterWarmUpJobLatestStartTime(String.valueOf(jobId),
srcClusterName,
- dstClusterName, startTimeMs);
- this.fetchBeToTabletIdBatches();
- long totalTablets = beToTabletIdBatches.values().stream()
- .flatMap(List::stream)
- .mapToLong(List::size)
- .sum();
- MetricRepo.increaseClusterWarmUpJobRequestedTablets(dstClusterName,
totalTablets);
- MetricRepo.increaseClusterWarmUpJobExecCount(dstClusterName);
+ long totalTablets;
+ try {
+ this.setJobDone = false;
+ this.lastBatchId = -1;
+ this.startTimeMs = System.currentTimeMillis();
+ // reset clients to ensure we have the latest BE info
+ this.beToThriftAddress = null;
+ this.beToClient = null;
+ this.beToAddr = null;
+
MetricRepo.updateClusterWarmUpJobLatestStartTime(String.valueOf(jobId),
srcClusterName,
+ dstClusterName, startTimeMs);
+ this.fetchBeToTabletIdBatches();
+ totalTablets = beToTabletIdBatches.values().stream()
+ .flatMap(List::stream)
+ .mapToLong(List::size)
+ .sum();
+
MetricRepo.increaseClusterWarmUpJobRequestedTablets(dstClusterName,
totalTablets);
+ MetricRepo.increaseClusterWarmUpJobExecCount(dstClusterName);
+ } catch (Exception e) {
+ LOG.warn("failed to initialize cloud warm up job {}", jobId, e);
+ // No BE job has started. Reuse cancellation to release the
destination registration
+ // and preserve periodic jobs for their next scheduled attempt.
+ cancel("Failed to initialize warm up job: " + e.getMessage(),
false);
+ return;
+ }
this.jobState = JobState.RUNNING;
Env.getCurrentEnv().getEditLog().logModifyCloudWarmUpJob(this);
LOG.info("warmup-lock state-transition jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/CloudWarmUpJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/CloudWarmUpJobTest.java
index 800dd89c6b2..9ace836e147 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/cloud/CloudWarmUpJobTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/CloudWarmUpJobTest.java
@@ -56,6 +56,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReference;
public class CloudWarmUpJobTest {
@@ -186,6 +187,104 @@ public class CloudWarmUpJobTest {
Mockito.verify(editLog).logModifyCloudWarmUpJob(job);
}
+ @Test
+ public void testOncePendingInitializationFailureReleasesDestinationLock()
throws Exception {
+
checkPendingInitializationFailureReleasesDestinationLock(SyncMode.ONCE);
+ }
+
+ @Test
+ public void
testPeriodicPendingInitializationFailureReleasesDestinationLock() throws
Exception {
+
checkPendingInitializationFailureReleasesDestinationLock(SyncMode.PERIODIC);
+ }
+
+ private void
checkPendingInitializationFailureReleasesDestinationLock(SyncMode syncMode)
throws Exception {
+ CloudWarmUpJob job = Mockito.spy(createPendingJob(204L, syncMode));
+ CloudWarmUpJob nextJob = createPendingJob(205L, SyncMode.ONCE);
+ CloudEnv cloudEnv = Mockito.mock(CloudEnv.class);
+ CacheHotspotManager manager = new
CacheHotspotManager(Mockito.mock(CloudSystemInfoService.class),
+ Mockito.mock(ThreadPoolExecutor.class));
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(cloudEnv.getCacheHotspotMgr()).thenReturn(manager);
+ Mockito.when(cloudEnv.getEditLog()).thenReturn(editLog);
+ Mockito.doAnswer(invocation -> {
+ Assert.assertFalse(manager.tryRegisterRunningJob(nextJob));
+ throw new IllegalStateException("initialization failed");
+ }).when(job).fetchBeToTabletIdBatches();
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(cloudEnv);
+ job.run();
+
+ Assert.assertTrue(manager.tryRegisterRunningJob(nextJob));
+ Assert.assertEquals(syncMode == SyncMode.ONCE ? JobState.CANCELLED
: JobState.PENDING,
+ job.getJobState());
+ Assert.assertEquals("Failed to initialize warm up job:
initialization failed", job.getErrMsg());
+ Assert.assertTrue(job.getStartTimeMs() > 0);
+ Assert.assertTrue(job.getFinishedTimeMs() >= job.getStartTimeMs());
+ Assert.assertEquals(syncMode == SyncMode.PERIODIC,
job.shouldWait());
+ Mockito.verify(editLog).logModifyCloudWarmUpJob(job);
+
+ CloudWarmUpJob persistedJob = copyBySerialization(job);
+ Assert.assertEquals(job.getJobState(), persistedJob.getJobState());
+ Assert.assertEquals(job.getErrMsg(), persistedJob.getErrMsg());
+ Assert.assertEquals(job.getStartTimeMs(),
persistedJob.getStartTimeMs());
+ Assert.assertEquals(job.getFinishedTimeMs(),
persistedJob.getFinishedTimeMs());
+
+ nextJob.run();
+ Assert.assertEquals(JobState.RUNNING, nextJob.getJobState());
+ Mockito.verifyNoInteractions(mockBackendPool);
+
+ if (syncMode == SyncMode.PERIODIC) {
+ manager.notifyJobStop(nextJob);
+
Mockito.doCallRealMethod().when(job).fetchBeToTabletIdBatches();
+ setStartTimeMs(job, System.currentTimeMillis() - 61_000L);
+ Assert.assertFalse(job.shouldWait());
+ job.run();
+ Assert.assertEquals(JobState.RUNNING, job.getJobState());
+ Assert.assertFalse(manager.tryRegisterRunningJob(nextJob));
+ } else {
+ job.run();
+ Mockito.verify(job).fetchBeToTabletIdBatches();
+ Assert.assertEquals(JobState.CANCELLED, job.getJobState());
+ }
+ }
+ }
+
+ @Test
+ public void testOncePendingInitializationKeepsDestinationLockOnSuccess() {
+ checkPendingInitializationKeepsDestinationLockOnSuccess(SyncMode.ONCE);
+ }
+
+ @Test
+ public void
testPeriodicPendingInitializationKeepsDestinationLockOnSuccess() {
+
checkPendingInitializationKeepsDestinationLockOnSuccess(SyncMode.PERIODIC);
+ }
+
+ private void
checkPendingInitializationKeepsDestinationLockOnSuccess(SyncMode syncMode) {
+ CloudWarmUpJob job = createPendingJob(206L, syncMode);
+ CloudWarmUpJob nextJob = Mockito.spy(createPendingJob(207L,
SyncMode.ONCE));
+ CloudEnv cloudEnv = Mockito.mock(CloudEnv.class);
+ CacheHotspotManager manager = new
CacheHotspotManager(Mockito.mock(CloudSystemInfoService.class),
+ Mockito.mock(ThreadPoolExecutor.class));
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(cloudEnv.getCacheHotspotMgr()).thenReturn(manager);
+ Mockito.when(cloudEnv.getEditLog()).thenReturn(editLog);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(cloudEnv);
+ job.run();
+ Assert.assertEquals(JobState.RUNNING, job.getJobState());
+
+ nextJob.run();
+ Assert.assertEquals(JobState.PENDING, nextJob.getJobState());
+ Assert.assertEquals(-1L, nextJob.getStartTimeMs());
+ Assert.assertFalse(manager.tryRegisterRunningJob(nextJob));
+ Mockito.verify(nextJob,
Mockito.never()).fetchBeToTabletIdBatches();
+ Mockito.verify(editLog,
Mockito.never()).logModifyCloudWarmUpJob(nextJob);
+ Mockito.verify(editLog).logModifyCloudWarmUpJob(job);
+ }
+ }
+
@Test
public void testEventDrivenSuccessfulRetryClearsErrMsg() throws Exception {
CloudSystemInfoService cloudSystemInfoService =
Mockito.mock(CloudSystemInfoService.class);
@@ -325,6 +424,17 @@ public class CloudWarmUpJobTest {
Mockito.verify(mockBackendPool).returnObject(address, client);
}
+ private CloudWarmUpJob createPendingJob(long jobId, SyncMode syncMode) {
+ return new CloudWarmUpJob.Builder()
+ .setJobId(jobId)
+ .setSrcClusterName("source_cluster")
+ .setDstClusterName("target_cluster")
+ .setJobType(JobType.CLUSTER)
+ .setSyncMode(syncMode)
+ .setSyncInterval(60L)
+ .build();
+ }
+
private CloudWarmUpJob createRunningJob(long jobId, TNetworkAddress
firstAddress,
TNetworkAddress secondAddress) {
CloudWarmUpJob job = new CloudWarmUpJob.Builder()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
