This is an automated email from the ASF dual-hosted git repository.
bobhan1 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 96db864bb07 [fix](cloud) Restore warm-up destination locks after FE
recovery (#67867)
96db864bb07 is described below
commit 96db864bb071aa41d96a9c491e00c1cbabf7dc15
Author: bobhan1 <[email protected]>
AuthorDate: Thu Sep 17 11:23:47 2026 +0800
[fix](cloud) Restore warm-up destination locks after FE recovery (#67867)
### What problem does this PR solve?
Problem Summary:
A Cloud ONCE/PERIODIC warm-up job restored as RUNNING after FE restart
or master failover bypasses the PENDING registration path. Its
destination is therefore absent from the in-memory running-job map,
allowing a newly submitted job for that destination to start and
encounter a BE job conflict.
Rebuild destination ownership from the final restored job states before
the master becomes ready and starts warm-up scheduling. Register RUNNING
non-event jobs through the existing `tryRegisterRunningJob()` method,
preserving their batch progress and leaving periodic jobs whose latest
journal state is PENDING unregistered. This covers both image loading
and journal replay.
Recovery only rebuilds in-memory ownership. It leaves job states
unchanged and introduces no cancellation, RPC, or journal writes.
Reconciliation of pre-existing conflicting RUNNING jobs is outside this
PR.
Add FE tests covering image/journal recovery, final periodic states,
legacy ONCE jobs, preserved progress, queued execution with mocked BE
RPCs, repeated recovery, and safe owner release.
### Release note
Fix Cloud warm-up destination ownership recovery after FE restart or
master failover so newly submitted jobs wait for a recovered running job
on the same destination.
### Check List (For Author)
- Test:
- [x] Unit Test: 39 tests passed, including 13 recovery cases, with zero
failures, errors, or skips. A focused pre-fix test on the master
baseline reproduced the missing reservation (1 test, 1 failure).
- Checkstyle: the check of all three changed files passed with zero
violations. The earlier full FE check found only two pre-existing JUnit
4 import violations in unchanged `cloud/rpc/VersionHelperTest.java`.
- Docker regression was not run.
- Behavior changed:
- [x] Yes. Restore destination reservations before scheduling.
- Does this need documentation?
- [x] 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/CacheHotspotManager.java | 14 ++
.../org/apache/doris/cloud/catalog/CloudEnv.java | 1 +
.../cloud/CacheHotspotManagerRecoveryTest.java | 272 +++++++++++++++++++++
3 files changed, 287 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
index 528837151e4..f2abe2463c7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
@@ -506,6 +506,20 @@ public class CacheHotspotManager extends MasterDaemon {
// Ensures that at most one job runs concurrently per destination cluster.
private Map<String, Long> clusterToRunningJobId = new
ConcurrentHashMap<>();
+ /**
+ * Rebuild the runtime owners after all image and journal records have
been restored, before
+ * this FE becomes ready or starts scheduling warm-up jobs. Only the final
RUNNING state owns
+ * a destination: a periodic job may have returned to PENDING in a later
journal record.
+ */
+ public void recoverRunningJobsBeforeStart() {
+ Preconditions.checkState(!startJobDaemon, "Warm-up recovery must
precede job scheduling");
+ clusterToRunningJobId.clear();
+ cloudWarmUpJobs.values().stream()
+ .filter(job -> !job.isEventDriven() && job.getJobState() ==
JobState.RUNNING)
+ .forEach(this::tryRegisterRunningJob);
+ LOG.info("restored warm-up owners for {} destinations",
clusterToRunningJobId.size());
+ }
+
/**
* Attempts to register a job as running for the given destination cluster.
* <p>
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
index b482ddf260e..85103b73487 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
@@ -174,6 +174,7 @@ public class CloudEnv extends Env {
cloudClusterCheck.start();
cloudTabletRebalancer.start();
if (Config.enable_fetch_cluster_cache_hotspot) {
+ cacheHotspotMgr.recoverRunningJobsBeforeStart();
cacheHotspotMgr.start();
}
upgradeMgr.start();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerRecoveryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerRecoveryTest.java
new file mode 100644
index 00000000000..80a952cf1e5
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerRecoveryTest.java
@@ -0,0 +1,272 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cloud;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.CloudWarmUpJob.JobState;
+import org.apache.doris.cloud.CloudWarmUpJob.JobType;
+import org.apache.doris.cloud.CloudWarmUpJob.SyncEvent;
+import org.apache.doris.cloud.CloudWarmUpJob.SyncMode;
+import org.apache.doris.cloud.catalog.CloudEnv;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.GenericPool;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TStatus;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TWarmUpTabletsRequest;
+import org.apache.doris.thrift.TWarmUpTabletsRequestType;
+import org.apache.doris.thrift.TWarmUpTabletsResponse;
+
+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.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class CacheHotspotManagerRecoveryTest {
+ private CacheHotspotManager manager;
+ private CloudEnv env;
+ private EditLog editLog;
+ private MockedStatic<Env> envMock;
+ private GenericPool<BackendService.Client> originalBackendPool;
+ private GenericPool<BackendService.Client> backendPool;
+ private BackendService.Client client;
+ private boolean originalRunningUnitTest;
+
+ @SuppressWarnings("unchecked")
+ @BeforeEach
+ public void setUp() throws Exception {
+ originalRunningUnitTest = FeConstants.runningUnitTest;
+ FeConstants.runningUnitTest = false;
+ originalBackendPool = ClientPool.backendPool;
+ backendPool = Mockito.mock(GenericPool.class);
+ ClientPool.backendPool = backendPool;
+ client = Mockito.mock(BackendService.Client.class);
+
Mockito.when(backendPool.borrowObject(Mockito.any(TNetworkAddress.class))).thenReturn(client);
+
Mockito.when(client.warmUpTablets(Mockito.any())).thenAnswer(invocation ->
response(0));
+
+ CloudSystemInfoService systemInfo =
Mockito.mock(CloudSystemInfoService.class);
+ Backend backend = new Backend(1L, "127.0.0.1", 9050);
+ backend.setBePort(9060);
+ Mockito.when(systemInfo.getBackendsByClusterName(Mockito.anyString()))
+ .thenReturn(Collections.singletonList(backend));
+ manager = new CacheHotspotManager(systemInfo,
Mockito.mock(ThreadPoolExecutor.class));
+ env = Mockito.mock(CloudEnv.class);
+ editLog = Mockito.mock(EditLog.class);
+ Mockito.when(env.getCacheHotspotMgr()).thenReturn(manager);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
Mockito.when(env.loadCloudWarmUpJob(Mockito.any(DataInputStream.class),
Mockito.anyLong()))
+ .thenCallRealMethod();
+ envMock = Mockito.mockStatic(Env.class);
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfo);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ envMock.close();
+ ClientPool.backendPool = originalBackendPool;
+ FeConstants.runningUnitTest = originalRunningUnitTest;
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = SyncMode.class, names = {"ONCE", "PERIODIC"})
+ public void testRunningJobRecoveredFromJournalBlocksNewJob(SyncMode mode)
throws Exception {
+ CloudWarmUpJob job = newJob(204L, mode, JobState.RUNNING);
+ job.setLastBatchId(3L);
+ CloudWarmUpJob restored = roundTrip(job);
+ manager.replayCloudWarmUpJob(restored);
+ manager.recoverRunningJobsBeforeStart();
+
+ Assertions.assertFalse(manager.tryRegisterRunningJob(newJob(205L,
SyncMode.ONCE, JobState.PENDING)),
+ "The recovered running job must keep the destination
reserved");
+ Assertions.assertEquals(JobState.RUNNING, restored.getJobState());
+ Assertions.assertEquals(job.startTimeMs, restored.startTimeMs);
+ Assertions.assertEquals(3L, restored.getLastBatchId());
+ Assertions.assertEquals(job.beToTabletIdBatches,
restored.beToTabletIdBatches);
+ Mockito.verifyNoInteractions(editLog, backendPool, client);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = SyncMode.class, names = {"ONCE", "PERIODIC"})
+ public void testRunningJobRecoveredOnlyFromImageBlocksNewJob(SyncMode
mode) throws Exception {
+ loadImage(newJob(204L, mode, JobState.RUNNING));
+ manager.recoverRunningJobsBeforeStart();
+
+ Assertions.assertFalse(manager.tryRegisterRunningJob(newJob(205L,
SyncMode.ONCE, JobState.PENDING)));
+ Mockito.verifyNoInteractions(editLog, backendPool, client);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = JobState.class, names = {"PENDING", "FINISHED",
"CANCELLED", "DELETED"})
+ public void testLatestJournalStateOverridesRunningImage(JobState
finalState) throws Exception {
+ CloudWarmUpJob job = newJob(204L, SyncMode.PERIODIC, JobState.RUNNING);
+ loadImage(job);
+ job.setJobState(finalState);
+ manager.replayCloudWarmUpJob(roundTrip(job));
+ manager.recoverRunningJobsBeforeStart();
+
+ Assertions.assertTrue(manager.tryRegisterRunningJob(newJob(205L,
SyncMode.ONCE, JobState.PENDING)));
+ Mockito.verifyNoInteractions(editLog, backendPool, client);
+ }
+
+ @Test
+ public void testRepeatedReplayAndOldCompletionDoNotReleaseNewOwner()
throws Exception {
+ CloudWarmUpJob first = newJob(204L, SyncMode.PERIODIC,
JobState.RUNNING);
+ manager.replayCloudWarmUpJob(roundTrip(first));
+ manager.replayCloudWarmUpJob(roundTrip(first));
+ first.setJobState(JobState.PENDING);
+ manager.replayCloudWarmUpJob(roundTrip(first));
+ CloudWarmUpJob second = newJob(205L, SyncMode.ONCE, JobState.RUNNING);
+ manager.replayCloudWarmUpJob(roundTrip(second));
+ manager.recoverRunningJobsBeforeStart();
+ manager.recoverRunningJobsBeforeStart();
+ manager.notifyJobStop(first);
+
+ Assertions.assertFalse(manager.tryRegisterRunningJob(first));
+ Assertions.assertTrue(manager.tryRegisterRunningJob(second));
+ Mockito.verifyNoInteractions(editLog, backendPool, client);
+ }
+
+ @Test
+ public void testLegacyOnceJobWithoutSyncModeReservesDestination() throws
Exception {
+ CloudWarmUpJob job = newJob(204L, SyncMode.ONCE, JobState.RUNNING);
+ job.syncMode = null;
+ loadImage(job);
+ manager.recoverRunningJobsBeforeStart();
+
+ Assertions.assertTrue(manager.getCloudWarmUpJob(204L).isOnce());
+ Assertions.assertFalse(manager.tryRegisterRunningJob(newJob(205L,
SyncMode.ONCE, JobState.PENDING)));
+ }
+
+ @Test
+ public void testEventDrivenAndDifferentDestinationsRemainIndependent()
throws Exception {
+ CloudWarmUpJob event = newJob(204L, SyncMode.EVENT_DRIVEN,
JobState.RUNNING);
+ loadImage(event);
+ CloudWarmUpJob running = newJob(205L, SyncMode.ONCE, JobState.RUNNING);
+ running.setCloudClusterName("another_target");
+ manager.replayCloudWarmUpJob(roundTrip(running));
+ manager.recoverRunningJobsBeforeStart();
+
+ Assertions.assertTrue(manager.tryRegisterRunningJob(newJob(206L,
SyncMode.ONCE, JobState.PENDING)));
+ Assertions.assertTrue(manager.tryRegisterRunningJob(event));
+ CloudWarmUpJob blocked = newJob(207L, SyncMode.ONCE, JobState.PENDING);
+ blocked.setCloudClusterName("another_target");
+ Assertions.assertFalse(manager.tryRegisterRunningJob(blocked));
+ Mockito.verifyNoInteractions(editLog, backendPool, client);
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = SyncMode.class, names = {"ONCE", "PERIODIC"})
+ public void testRecoveredJobFinishesBeforeQueuedJobStarts(SyncMode mode)
throws Exception {
+ CloudWarmUpJob running = newJob(204L, mode, JobState.RUNNING);
+ running.setLastBatchId(0L);
+ running.setBeToTabletIdBatches(Collections.singletonMap(1L,
+ Arrays.asList(Collections.singletonList(11L),
Collections.singletonList(12L))));
+ manager.replayCloudWarmUpJob(roundTrip(running));
+ CloudWarmUpJob restored = manager.getCloudWarmUpJob(204L);
+ CloudWarmUpJob pending = newJob(205L, SyncMode.ONCE, JobState.PENDING);
+ pending.setJobType(JobType.TABLE);
+ manager.addCloudWarmUpJob(pending);
+ manager.recoverRunningJobsBeforeStart();
+ AtomicBoolean oldBatchPending = new AtomicBoolean(true);
+
Mockito.when(client.warmUpTablets(Mockito.any())).thenAnswer(invocation -> {
+ TWarmUpTabletsRequest request = invocation.getArgument(0);
+ return response(request.getType() ==
TWarmUpTabletsRequestType.GET_CURRENT_JOB_STATE_AND_LEASE
+ && oldBatchPending.get() ? 1 : 0);
+ });
+
+ restored.run();
+ pending.run();
+ Assertions.assertEquals(JobState.RUNNING, restored.getJobState());
+ Assertions.assertEquals(JobState.PENDING, pending.getJobState());
+ Mockito.verify(client,
Mockito.never()).warmUpTablets(Mockito.argThat(request -> request.getJobId() ==
205L));
+ Mockito.verifyNoInteractions(editLog);
+
+ oldBatchPending.set(false);
+ restored.run();
+ Assertions.assertEquals(mode == SyncMode.ONCE ? JobState.FINISHED :
JobState.PENDING,
+ restored.getJobState());
+ pending.run();
+ Assertions.assertEquals(JobState.RUNNING, pending.getJobState());
+ pending.run();
+ pending.run();
+ Assertions.assertEquals(JobState.FINISHED, pending.getJobState());
+ Assertions.assertEquals("", pending.errMsg);
+ Mockito.verify(client).warmUpTablets(Mockito.argThat(request ->
request.getJobId() == 205L
+ && request.getType() == TWarmUpTabletsRequestType.SET_JOB));
+ }
+
+ private void loadImage(CloudWarmUpJob job) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes);
+ output.writeInt(0); // Legacy runnable jobs.
+ output.writeInt(0); // Legacy finished jobs.
+ output.writeInt(1);
+ job.write(output);
+ Assertions.assertEquals(1L, env.loadCloudWarmUpJob(
+ new DataInputStream(new
ByteArrayInputStream(bytes.toByteArray())), 0L));
+ }
+
+ private CloudWarmUpJob roundTrip(CloudWarmUpJob job) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ job.write(new DataOutputStream(bytes));
+ return CloudWarmUpJob.read(new DataInputStream(new
ByteArrayInputStream(bytes.toByteArray())));
+ }
+
+ private CloudWarmUpJob newJob(long jobId, SyncMode syncMode, JobState
state) {
+ CloudWarmUpJob job = new CloudWarmUpJob.Builder()
+ .setJobId(jobId)
+ .setSrcClusterName("source_" + jobId)
+ .setDstClusterName("target_cluster")
+ .setSyncMode(syncMode)
+ .setSyncEvent(SyncEvent.LOAD)
+ .setSyncInterval(60L)
+ .build();
+ job.setJobState(state);
+ job.startTimeMs = System.currentTimeMillis();
+ job.setBeToThriftAddress(Collections.singletonMap(1L,
"127.0.0.1:9060"));
+ job.setBeToTabletIdBatches(Collections.singletonMap(1L,
+ Collections.singletonList(Collections.singletonList(11L))));
+ return job;
+ }
+
+ private TWarmUpTabletsResponse response(int pendingJobs) {
+ TWarmUpTabletsResponse response = new TWarmUpTabletsResponse();
+ response.setStatus(new TStatus(TStatusCode.OK));
+ response.setPendingJobSize(pendingJobs);
+ return response;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
