This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 f0affa953df [fix](cloud) Release warm-up destination on initialization
failure (#67924)
f0affa953df is described below
commit f0affa953dff3d0e761bd6bb2f62b231b31a9e26
Author: bobhan1 <[email protected]>
AuthorDate: Thu Sep 17 11:22:26 2026 +0800
[fix](cloud) Release warm-up destination on initialization failure (#67924)
A cloud warm-up job registers its destination compute group before
initializing tablet batches. If initialization throws, the outer `run()`
handler only logs the exception: the job stays `PENDING` and keeps the
destination registration. Other ONCE/PERIODIC jobs targeting that group
cannot start. A later successful retry can recover the original job, but
repeated initialization failures can block the group indefinitely
because the warm-up timeout only applies to `RUNNING` jobs.
Catch initialization failures before transitioning to `RUNNING` and
reuse `cancel(..., false)` to persist the error and release the
destination registration. ONCE jobs become `CANCELLED`; PERIODIC jobs
remain `PENDING` and retry at their existing interval. Initialization
has not submitted work to BEs, so this path does not send cleanup RPCs.
Successful initialization retains the destination registration as
before.
### Release note
Release the destination compute group when cloud warm-up initialization
fails, allowing subsequent warm-up jobs to proceed. Report the
initialization error and preserve periodic retry scheduling.
---
.../org/apache/doris/cloud/CloudWarmUpJob.java | 42 ++++++----
.../org/apache/doris/cloud/CloudWarmUpJobTest.java | 96 ++++++++++++++++++++++
2 files changed, 121 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 dce7f8e675d..f54fd9dfba3 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
@@ -41,6 +41,8 @@ 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.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -56,6 +58,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 +189,88 @@ public class CloudWarmUpJobTest {
Mockito.verify(editLog).logModifyCloudWarmUpJob(job);
}
+ @ParameterizedTest
+ @EnumSource(value = SyncMode.class, names = {"ONCE", "PERIODIC"})
+ public void
testPendingInitializationFailureReleasesDestinationLock(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 -> {
+ Assertions.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();
+
+ Assertions.assertTrue(manager.tryRegisterRunningJob(nextJob));
+ Assertions.assertEquals(syncMode == SyncMode.ONCE ?
JobState.CANCELLED : JobState.PENDING,
+ job.getJobState());
+ Assertions.assertEquals("Failed to initialize warm up job:
initialization failed", job.getErrMsg());
+ Assertions.assertTrue(job.getStartTimeMs() > 0);
+ Assertions.assertTrue(job.getFinishedTimeMs() >=
job.getStartTimeMs());
+ Assertions.assertEquals(syncMode == SyncMode.PERIODIC,
job.shouldWait());
+ Mockito.verify(editLog).logModifyCloudWarmUpJob(job);
+
+ CloudWarmUpJob persistedJob = copyBySerialization(job);
+ Assertions.assertEquals(job.getJobState(),
persistedJob.getJobState());
+ Assertions.assertEquals(job.getErrMsg(), persistedJob.getErrMsg());
+ Assertions.assertEquals(job.getStartTimeMs(),
persistedJob.getStartTimeMs());
+ Assertions.assertEquals(job.getFinishedTimeMs(),
persistedJob.getFinishedTimeMs());
+
+ nextJob.run();
+ Assertions.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);
+ Assertions.assertFalse(job.shouldWait());
+ job.run();
+ Assertions.assertEquals(JobState.RUNNING, job.getJobState());
+ Assertions.assertFalse(manager.tryRegisterRunningJob(nextJob));
+ } else {
+ job.run();
+ Mockito.verify(job).fetchBeToTabletIdBatches();
+ Assertions.assertEquals(JobState.CANCELLED, job.getJobState());
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = SyncMode.class, names = {"ONCE", "PERIODIC"})
+ public void
testPendingInitializationKeepsDestinationLockOnSuccess(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();
+ Assertions.assertEquals(JobState.RUNNING, job.getJobState());
+
+ nextJob.run();
+ Assertions.assertEquals(JobState.PENDING, nextJob.getJobState());
+ Assertions.assertEquals(-1L, nextJob.getStartTimeMs());
+ Assertions.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 +410,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]
