branch-4.1: [fix](cloud) Release warm-up destination on initialization failure #67924

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={} "

fe/fe-core/src/test/java/org/apache/doris/cloud/CloudWarmUpJobTest.java

@@ -56,6 +56,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 +187,104 @@ public void testPendingRetryKeepsErrMsgWhenJobStarts() throws Exception {
         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 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()