From 0b7a7a4df86714031b866b882841c11655abe436 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Mon, 14 Sep 2026 10:56:54 +0800 Subject: [PATCH] [fix](cloud) Release warm-up destination on initialization failure ### What problem does this PR solve? Problem Summary: 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. ### Check List (For Author) - Test: Unit Test. All 30 tests passed across `CloudWarmUpJobTest` (11), `CacheHotspotManagerSchedulerTest` (4), and `cloud.cache.CacheHotspotManagerTest` (15). New tests cover ONCE/PERIODIC initialization failure using the real destination registration map, persisted failure state, periodic retry, and continued mutual exclusion after successful initialization. Before the fix, both failure-injection cases failed at the subsequent-job registration assertion; both successful-initialization cases passed. - Validation commands: `./build.sh --fe -j100` passed, including Checkstyle; `./run-fe-ut.sh --run 'org.apache.doris.cloud.CloudWarmUpJobTest,org.apache.doris.cloud.CacheHotspotManagerSchedulerTest,org.apache.doris.cloud.cache.CacheHotspotManagerTest'` passed; `git diff --check` passed. No live cloud-cluster SQL regression was run. - Behavior changed: Yes. Failed ONCE initialization cancels the job; failed PERIODIC initialization releases the destination registration and waits for its next interval. - Does this need documentation: No. ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label --- .../apache/doris/cloud/CloudWarmUpJob.java | 42 ++++---- .../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 e2f58cc49b04ab..6fbe6188d5e378 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 @@ private void runPendingJob() throws DdlException { 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 dce7f8e675d105..f54fd9dfba37b8 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.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.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 void testPendingRetryKeepsErrMsgWhenJobStarts() throws Exception { 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 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 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 void testRunningRetryClearsErrMsgWhenJobFinishes() throws Exception { 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()