Skip to content

Commit 94ad62a

Browse files
bobhan1bobhan1
authored
[fix](cloud) prevent warm-up job scheduling starvation (#67527)
Problem Summary: When cloud warm-up jobs outnumber the configured active slots, the scheduler can over-submit tasks because a job is recorded as active only after its worker starts. The direct-handoff thread pool previously discarded rejected tasks, and unordered map traversal could repeatedly favor the same jobs, leaving other runnable jobs pending indefinitely. This PR keeps the direct-handoff pool and makes scheduling bounded and retryable: - Scan runnable jobs once and keep only the jobs needed by the current slots in a bounded priority heap. - Reserve the active slot before submission and roll it back when submission is rejected. - Schedule the least-recently-run jobs first, with never-scheduled one-time jobs taking the first turn. - Surface pool rejection so the scheduler can retry the job in the next cycle. - Apply mutable scheduler concurrency and interval configuration without requiring an FE restart. ### Release note Fix cloud warm-up jobs remaining pending when the scheduler thread pool is saturated.
1 parent c275e05 commit 94ad62a

4 files changed

Lines changed: 282 additions & 22 deletions

File tree

‎fe/fe-common/src/main/java/org/apache/doris/common/Config.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3345,9 +3345,21 @@ public static int metaServiceRpcRetryTimes() {
33453345
@ConfField(mutable = true, masterOnly = true)
33463346
public static int cloud_warm_up_timeout_second = 86400 * 30; // 30 days
33473347

3348-
@ConfField(mutable = true, masterOnly = true)
3348+
@ConfField(mutable = true, masterOnly = true,
3349+
callback = PositiveCloudWarmUpSchedulerIntervalConfHandler.class)
33493350
public static int cloud_warm_up_job_scheduler_interval_millisecond = 1000; // 1 seconds
33503351

3352+
public static class PositiveCloudWarmUpSchedulerIntervalConfHandler implements ConfHandler {
3353+
@Override
3354+
public void handle(Field field, String value) throws Exception {
3355+
int parsedValue = Integer.parseInt(value.trim());
3356+
if (parsedValue <= 0) {
3357+
throw new ConfigException(field.getName() + " must be greater than 0");
3358+
}
3359+
field.setInt(null, parsedValue);
3360+
}
3361+
}
3362+
33513363
@ConfField(mutable = true, masterOnly = true)
33523364
public static long cloud_warm_up_job_max_bytes_per_batch = 21474836480L; // 20GB
33533365

‎fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,27 @@ public void testSetWebSqlMaxResultBytes() throws ConfigException {
165165
}
166166
}
167167

168+
@Test
169+
public void testCloudWarmUpSchedulerIntervalMustBePositive() throws ConfigException {
170+
int original = Config.cloud_warm_up_job_scheduler_interval_millisecond;
171+
try {
172+
ConfigBase.setMutableConfig("cloud_warm_up_job_scheduler_interval_millisecond", "2000");
173+
Assert.assertEquals(2000, Config.cloud_warm_up_job_scheduler_interval_millisecond);
174+
175+
ConfigException zeroException = Assert.assertThrows(ConfigException.class,
176+
() -> ConfigBase.setMutableConfig("cloud_warm_up_job_scheduler_interval_millisecond", "0"));
177+
Assert.assertTrue(zeroException.getMessage().contains("must be greater than 0"));
178+
Assert.assertEquals(2000, Config.cloud_warm_up_job_scheduler_interval_millisecond);
179+
180+
ConfigException negativeException = Assert.assertThrows(ConfigException.class,
181+
() -> ConfigBase.setMutableConfig("cloud_warm_up_job_scheduler_interval_millisecond", "-1"));
182+
Assert.assertTrue(negativeException.getMessage().contains("must be greater than 0"));
183+
Assert.assertEquals(2000, Config.cloud_warm_up_job_scheduler_interval_millisecond);
184+
} finally {
185+
Config.cloud_warm_up_job_scheduler_interval_millisecond = original;
186+
}
187+
}
188+
168189
@Test
169190
public void testValidateWebSqlStartupConfig() throws ConfigException {
170191
int originalIdleTimeout = Config.web_sql_session_idle_timeout_seconds;

‎fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java‎

Lines changed: 99 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@
8686
import java.util.Map.Entry;
8787
import java.util.Objects;
8888
import java.util.Optional;
89+
import java.util.PriorityQueue;
8990
import java.util.Set;
9091
import java.util.concurrent.CompletableFuture;
9192
import java.util.concurrent.ConcurrentHashMap;
@@ -95,9 +96,11 @@
9596
import java.util.concurrent.ExecutorService;
9697
import java.util.concurrent.Executors;
9798
import java.util.concurrent.Future;
99+
import java.util.concurrent.RejectedExecutionException;
98100
import java.util.concurrent.ThreadPoolExecutor;
99101
import java.util.concurrent.TimeUnit;
100102
import java.util.concurrent.TimeoutException;
103+
import java.util.concurrent.atomic.AtomicLong;
101104
import java.util.concurrent.locks.ReentrantLock;
102105
import java.util.stream.Collectors;
103106

@@ -139,11 +142,16 @@ public class CacheHotspotManager extends MasterDaemon {
139142

140143
private ConcurrentMap<Long, CloudWarmUpJob> runnableCloudWarmUpJobs = Maps.newConcurrentMap();
141144

145+
// Keep scheduling history in memory only. Jobs that have never been scheduled have sequence 0
146+
// and are selected before jobs that ran in previous cycles.
147+
private final ConcurrentMap<Long, Long> cloudWarmUpJobLastScheduleSeq = Maps.newConcurrentMap();
148+
149+
private final AtomicLong cloudWarmUpJobScheduleSeq = new AtomicLong(0);
150+
142151
private final ConcurrentMap<OncePendingJobKey, RefCountedPendingCreateLock> oncePendingCreateLocks
143152
= Maps.newConcurrentMap();
144153

145-
private final ThreadPoolExecutor cloudWarmUpThreadPool = ThreadPoolManager.newDaemonCacheThreadPool(
146-
Config.max_active_cloud_warm_up_job, "cloud-warm-up-pool", true);
154+
private final ThreadPoolExecutor cloudWarmUpThreadPool;
147155

148156
private static class JobKey {
149157
private final String srcName;
@@ -593,8 +601,15 @@ public void notifyJobStop(CloudWarmUpJob job) {
593601
}
594602

595603
public CacheHotspotManager(CloudSystemInfoService nodeMgr) {
604+
this(nodeMgr, ThreadPoolManager.newDaemonCacheThreadPoolThrowException(
605+
Config.max_active_cloud_warm_up_job, "cloud-warm-up-pool", true));
606+
}
607+
608+
@VisibleForTesting
609+
CacheHotspotManager(CloudSystemInfoService nodeMgr, ThreadPoolExecutor cloudWarmUpThreadPool) {
596610
super("CacheHotspotManager", Config.fetch_cluster_cache_hotspot_interval_ms);
597611
this.nodeMgr = nodeMgr;
612+
this.cloudWarmUpThreadPool = cloudWarmUpThreadPool;
598613
}
599614

600615
@Override
@@ -1003,6 +1018,10 @@ private class JobDaemon extends MasterDaemon {
10031018

10041019
@Override
10051020
public void runAfterCatalogReady() {
1021+
if (getInterval() != Config.cloud_warm_up_job_scheduler_interval_millisecond) {
1022+
setInterval(Config.cloud_warm_up_job_scheduler_interval_millisecond);
1023+
LOG.info("update cloud warm up job daemon interval to {}ms", getInterval());
1024+
}
10061025
if (cycleCount >= CYCLE_COUNT_TO_CHECK_EXPIRE_CLOUD_WARM_UP_JOB) {
10071026
clearFinishedOrCancelCloudWarmUpJob();
10081027
cycleCount = 0;
@@ -1202,6 +1221,7 @@ private void clearFinishedOrCancelCloudWarmUpJob() {
12021221
CloudWarmUpJob cloudWarmUpJob = iterator.next().getValue();
12031222
if (cloudWarmUpJob.isDone()) {
12041223
iterator.remove();
1224+
cloudWarmUpJobLastScheduleSeq.remove(cloudWarmUpJob.getJobId());
12051225
}
12061226
}
12071227
Iterator<Map.Entry<Long, CloudWarmUpJob>> iterator2 = cloudWarmUpJobs.entrySet().iterator();
@@ -1504,28 +1524,86 @@ public void cancelTableFilterJobsForClusterChange(String clusterName, String rea
15041524
}
15051525
}
15061526

1507-
private void runCloudWarmUpJob() {
1508-
runnableCloudWarmUpJobs.values().forEach(cloudWarmUpJob -> {
1509-
if (cloudWarmUpJob.shouldWait()) {
1510-
return;
1527+
@VisibleForTesting
1528+
void runCloudWarmUpJob() {
1529+
int maxActiveJobs = Config.max_active_cloud_warm_up_job;
1530+
if (maxActiveJobs <= 0) {
1531+
return;
1532+
}
1533+
1534+
if (cloudWarmUpThreadPool.getMaximumPoolSize() != maxActiveJobs) {
1535+
cloudWarmUpThreadPool.setMaximumPoolSize(maxActiveJobs);
1536+
LOG.info("resize cloud warm up thread pool to {}", maxActiveJobs);
1537+
}
1538+
1539+
int availableSlots = maxActiveJobs - activeCloudWarmUpJobs.size();
1540+
if (availableSlots <= 0) {
1541+
return;
1542+
}
1543+
1544+
// A smaller last-scheduled sequence has higher priority, and never-scheduled jobs use sequence 0.
1545+
// For the same sequence, prefer ONCE jobs, then earlier creation time, then smaller job ID.
1546+
Comparator<CloudWarmUpJob> schedulePriority = Comparator
1547+
.comparingLong((CloudWarmUpJob job) ->
1548+
cloudWarmUpJobLastScheduleSeq.getOrDefault(job.getJobId(), 0L))
1549+
.thenComparingInt(job -> job.isOnce() ? 0 : 1)
1550+
.thenComparingLong(CloudWarmUpJob::getCreateTimeMs)
1551+
.thenComparingLong(CloudWarmUpJob::getJobId);
1552+
1553+
// Keep only the highest-priority jobs needed by this cycle. The reversed comparator keeps
1554+
// the lowest-priority selected job at the heap top so it can be replaced during the scan.
1555+
PriorityQueue<CloudWarmUpJob> candidates = new PriorityQueue<>(schedulePriority.reversed());
1556+
for (CloudWarmUpJob job : runnableCloudWarmUpJobs.values()) {
1557+
if (job.shouldWait() || job.isDone() || activeCloudWarmUpJobs.containsKey(job.getJobId())) {
1558+
continue;
15111559
}
1512-
if (!cloudWarmUpJob.isDone() && !activeCloudWarmUpJobs.containsKey(cloudWarmUpJob.getJobId())
1513-
&& activeCloudWarmUpJobs.size() < Config.max_active_cloud_warm_up_job) {
1514-
if (FeConstants.runningUnitTest) {
1515-
cloudWarmUpJob.run();
1516-
} else {
1517-
cloudWarmUpThreadPool.submit(() -> {
1518-
if (activeCloudWarmUpJobs.putIfAbsent(cloudWarmUpJob.getJobId(), cloudWarmUpJob) == null) {
1519-
try {
1520-
cloudWarmUpJob.run();
1521-
} finally {
1522-
activeCloudWarmUpJobs.remove(cloudWarmUpJob.getJobId());
1523-
}
1524-
}
1525-
});
1560+
if (candidates.size() < availableSlots) {
1561+
candidates.offer(job);
1562+
} else if (schedulePriority.compare(job, candidates.peek()) < 0) {
1563+
candidates.poll();
1564+
candidates.offer(job);
1565+
}
1566+
}
1567+
1568+
List<CloudWarmUpJob> jobsToSchedule = new ArrayList<>(candidates);
1569+
jobsToSchedule.sort(schedulePriority);
1570+
1571+
for (CloudWarmUpJob job : jobsToSchedule) {
1572+
if (availableSlots <= 0) {
1573+
break;
1574+
}
1575+
long jobId = job.getJobId();
1576+
if (activeCloudWarmUpJobs.putIfAbsent(jobId, job) != null) {
1577+
continue;
1578+
}
1579+
1580+
if (FeConstants.runningUnitTest) {
1581+
cloudWarmUpJobLastScheduleSeq.put(jobId, cloudWarmUpJobScheduleSeq.incrementAndGet());
1582+
try {
1583+
job.run();
1584+
} finally {
1585+
activeCloudWarmUpJobs.remove(jobId, job);
15261586
}
1587+
--availableSlots;
1588+
continue;
15271589
}
1528-
});
1590+
1591+
try {
1592+
cloudWarmUpThreadPool.execute(() -> {
1593+
try {
1594+
job.run();
1595+
} finally {
1596+
activeCloudWarmUpJobs.remove(jobId, job);
1597+
}
1598+
});
1599+
cloudWarmUpJobLastScheduleSeq.put(jobId, cloudWarmUpJobScheduleSeq.incrementAndGet());
1600+
--availableSlots;
1601+
} catch (RejectedExecutionException e) {
1602+
activeCloudWarmUpJobs.remove(jobId, job);
1603+
LOG.warn("failed to schedule cloud warm up job {}, retry in next cycle", jobId);
1604+
break;
1605+
}
1606+
}
15291607
}
15301608

15311609
public void replayCloudWarmUpJob(CloudWarmUpJob cloudWarmUpJob) throws Exception {

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

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.cloud;
19+
20+
import org.apache.doris.cloud.system.CloudSystemInfoService;
21+
import org.apache.doris.common.Config;
22+
import org.apache.doris.common.FeConstants;
23+
24+
import org.junit.After;
25+
import org.junit.Assert;
26+
import org.junit.Before;
27+
import org.junit.Test;
28+
import org.mockito.Mockito;
29+
30+
import java.util.ArrayList;
31+
import java.util.Arrays;
32+
import java.util.List;
33+
import java.util.concurrent.RejectedExecutionException;
34+
import java.util.concurrent.ThreadPoolExecutor;
35+
import java.util.concurrent.atomic.AtomicInteger;
36+
37+
public class CacheHotspotManagerSchedulerTest {
38+
private boolean originalRunningUnitTest;
39+
private int originalMaxActiveCloudWarmUpJob;
40+
private ThreadPoolExecutor executor;
41+
private CacheHotspotManager manager;
42+
43+
@Before
44+
public void setUp() {
45+
originalRunningUnitTest = FeConstants.runningUnitTest;
46+
originalMaxActiveCloudWarmUpJob = Config.max_active_cloud_warm_up_job;
47+
FeConstants.runningUnitTest = false;
48+
Config.max_active_cloud_warm_up_job = 2;
49+
50+
executor = Mockito.mock(ThreadPoolExecutor.class);
51+
Mockito.when(executor.getMaximumPoolSize()).thenReturn(2);
52+
manager = new CacheHotspotManager(Mockito.mock(CloudSystemInfoService.class), executor);
53+
}
54+
55+
@After
56+
public void tearDown() {
57+
FeConstants.runningUnitTest = originalRunningUnitTest;
58+
Config.max_active_cloud_warm_up_job = originalMaxActiveCloudWarmUpJob;
59+
}
60+
61+
@Test
62+
public void testNewOnceJobGetsFirstTurnAndJobsRotate() throws Exception {
63+
List<Long> runOrder = new ArrayList<>();
64+
Mockito.doAnswer(invocation -> {
65+
((Runnable) invocation.getArgument(0)).run();
66+
return null;
67+
}).when(executor).execute(Mockito.any(Runnable.class));
68+
69+
manager.addCloudWarmUpJob(mockJob(1L, false, 1L, runOrder));
70+
manager.addCloudWarmUpJob(mockJob(2L, false, 2L, runOrder));
71+
manager.addCloudWarmUpJob(mockJob(3L, false, 3L, runOrder));
72+
manager.addCloudWarmUpJob(mockJob(4L, true, 4L, runOrder));
73+
74+
manager.runCloudWarmUpJob();
75+
manager.runCloudWarmUpJob();
76+
manager.runCloudWarmUpJob();
77+
manager.runCloudWarmUpJob();
78+
79+
Assert.assertEquals(Arrays.asList(4L, 1L, 2L, 3L, 4L, 1L, 2L, 3L), runOrder);
80+
}
81+
82+
@Test
83+
public void testActiveJobIsNotSubmittedAgain() throws Exception {
84+
List<Runnable> submittedTasks = new ArrayList<>();
85+
Mockito.doAnswer(invocation -> {
86+
submittedTasks.add(invocation.getArgument(0));
87+
return null;
88+
}).when(executor).execute(Mockito.any(Runnable.class));
89+
90+
CloudWarmUpJob job = mockJob(1L, true, 1L, new ArrayList<>());
91+
manager.addCloudWarmUpJob(job);
92+
93+
manager.runCloudWarmUpJob();
94+
manager.runCloudWarmUpJob();
95+
Assert.assertEquals(1, submittedTasks.size());
96+
97+
submittedTasks.get(0).run();
98+
manager.runCloudWarmUpJob();
99+
Assert.assertEquals(2, submittedTasks.size());
100+
submittedTasks.get(1).run();
101+
Mockito.verify(job, Mockito.times(2)).run();
102+
}
103+
104+
@Test
105+
public void testRejectedJobIsRetried() throws Exception {
106+
AtomicInteger submitCount = new AtomicInteger();
107+
Mockito.doAnswer(invocation -> {
108+
if (submitCount.incrementAndGet() == 1) {
109+
throw new RejectedExecutionException("injected rejection");
110+
}
111+
((Runnable) invocation.getArgument(0)).run();
112+
return null;
113+
}).when(executor).execute(Mockito.any(Runnable.class));
114+
115+
CloudWarmUpJob job = mockJob(1L, true, 1L, new ArrayList<>());
116+
manager.addCloudWarmUpJob(job);
117+
118+
manager.runCloudWarmUpJob();
119+
Mockito.verify(job, Mockito.never()).run();
120+
manager.runCloudWarmUpJob();
121+
122+
Assert.assertEquals(2, submitCount.get());
123+
Mockito.verify(job, Mockito.times(1)).run();
124+
}
125+
126+
@Test
127+
public void testThreadPoolSizeFollowsMutableConfig() {
128+
Config.max_active_cloud_warm_up_job = 3;
129+
Mockito.when(executor.getMaximumPoolSize()).thenReturn(1);
130+
131+
manager.runCloudWarmUpJob();
132+
133+
Mockito.verify(executor).setMaximumPoolSize(3);
134+
}
135+
136+
private CloudWarmUpJob mockJob(long jobId, boolean once, long createTimeMs, List<Long> runOrder) {
137+
CloudWarmUpJob job = Mockito.mock(CloudWarmUpJob.class);
138+
Mockito.when(job.getJobId()).thenReturn(jobId);
139+
Mockito.when(job.isOnce()).thenReturn(once);
140+
Mockito.when(job.getCreateTimeMs()).thenReturn(createTimeMs);
141+
Mockito.when(job.shouldWait()).thenReturn(false);
142+
Mockito.when(job.isDone()).thenReturn(false);
143+
Mockito.doAnswer(invocation -> {
144+
runOrder.add(jobId);
145+
return null;
146+
}).when(job).run();
147+
return job;
148+
}
149+
}

0 commit comments

Comments
 (0)