github-actions[bot] commented on code in PR #67867:
URL: https://github.com/apache/doris/pull/67867#discussion_r4002741166
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java:
##########
@@ -506,6 +506,20 @@ private static String
formatExistingTableFilter(CloudWarmUpJob job) {
// 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);
Review Comment:
A failed registration here cannot be ignored. A finishing job releases this
map before journaling its PENDING/FINISHED state
(`CloudWarmUpJob.runRunningJob()`), so another job can journal RUNNING and a
crash in between legitimately leaves two RUNNING records for one destination.
This scan registers one but leaves the loser in `runnableCloudWarmUpJobs`; the
scheduler runs restored RUNNING jobs without checking ownership, even though BE
accepts only one job ID. If the registered job stops first, the destination is
exposed while the unowned job is still active. Please reconcile or park
conflicting restored jobs (or fail recovery loudly) before `JobDaemon` starts,
and cover the two-RUNNING case.
##########
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 -> {
Review Comment:
Please also cover the pre-send recovery state here. `runRunningJob()`
journals `lastBatchId++` before it sends the next `SET_BATCH`, so FE can fail
with cursor 0 persisted while BE still reports job 204/batch 0 and `pending=0`.
This mock's initial `pending=1` assumes batch 1 already arrived; it cannot
catch that other state. On replay FE currently ignores BE's returned job/batch
IDs, advances 0 to 1, requests nonexistent batch 2, and finishes without ever
submitting batch 1 (`[12]`). Please reconcile the cursors and assert that this
crash window still dispatches `SET_BATCH(1)` before releasing the owner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
