You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "anshul98ks123 (via GitHub)" <gi...@apache.org> on 2026/09/24 13:24:16 UTC
[PR] Honor concurrent scheduling in ad-hoc createTask [pinot]
anshul98ks123 opened a new pull request, #19660:
URL: https://github.com/apache/pinot/pull/19660
## Summary
Ad-hoc `PinotTaskManager#createTask` is the one task-generation entry point that still takes the controller-wide monitor unconditionally. With `controller.task.concurrentSchedulingEnabled`, cron generation for different tables already runs in parallel, but an ad-hoc request for one table still queues behind any other table's ad-hoc generation.
That matters because some generators do their expensive work inside it. A generator that lists an input prefix can hold the monitor for minutes on a large prefix, and every other ad-hoc request on the controller — for unrelated tables — waits the whole time, often past the client's timeout, without doing any work of its own.
## Change
`createTask` now dispatches the way `scheduleTasks` already does:
- **Concurrent path** — when every table the request targets resolves to concurrent scheduling (the `TableTaskConfig#concurrentSchedulingEnabled` override, else the cluster default) **and** distributed locking is enabled, the call holds no controller-wide monitor.
- **Legacy path** — otherwise, it runs under `synchronized(this)` exactly as before.
Both paths share `doCreateTask`, which is the previous method body unchanged. The `TableTaskConfig` docs now note that the flag covers ad-hoc creation as well as cron scheduling.
### Why distributed locking is required here
`shouldUseConcurrentPath` only recommends distributed locking. For ad-hoc creation it is required: once the monitor is gone, the per-table ZK lock is the only thing keeping two ad-hoc requests for the **same** table from generating tasks side by side. Without it the monitor stays, so same-table exclusion is never weaker than today.
### What runs without the monitor
Nothing new. Every step `doCreateTask` takes — `prepTaskQueue`, `isTaskSchedulable`, `acquireTaskLock`, `generateTasks`, `validatePinotTaskConfigs`, `addDefaultsToTaskConfig`, `submitTasks` — is already run without the monitor by the concurrent cron path in `doScheduleTasks`. This change adds a caller, not a new unlocked step.
## Backward compatibility
No behaviour change by default: both `controller.task.concurrentSchedulingEnabled` and `controller.task.enableDistributedLocking` default to `false`, so every request takes the legacy path. An unresolvable table name also takes the legacy path and fails there exactly as it did before.
## Testing
Nine new cases in `PinotTaskManagerConcurrentSchedulingTest`:
- **Resolver** — table opt-in, cluster default inherited, table opt-out over cluster default, no distributed locking, hybrid table with one type opted out, unknown table, missing table config.
- **Behaviour** — table A's generation parks inside `createTask`; a request for table B then either proceeds (both opted in) or is `BLOCKED` on the monitor (not opted in).
The opted-in behavioural test fails against the previous code — *"table B waited on table A although both opted into concurrent scheduling"* — and passes with this change. `spotless`, `checkstyle` and `license` are clean on `pinot-spi` and `pinot-controller`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Honor concurrent scheduling in ad-hoc createTask [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19660:
URL: https://github.com/apache/pinot/pull/19660#discussion_r4311662945
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java:
##########
@@ -235,7 +236,56 @@ public void init() {
}
}
- public synchronized Map<String, String> createTask(String taskType, String tableName, @Nullable String taskName,
+ /// Creates ad-hoc tasks for the given table.
+ ///
+ /// Dispatches like [#scheduleTasks]: when every targeted table opts into concurrent scheduling (see
+ /// [#shouldUseConcurrentPathForAdhoc]) the call holds no controller-wide monitor, so ad-hoc generation for
+ /// one table no longer waits behind another's; otherwise it runs under `synchronized(this)` as before. Both
+ /// paths share [#doCreateTask].
+ public Map<String, String> createTask(String taskType, String tableName, @Nullable String taskName,
+ Map<String, String> taskConfigs)
+ throws Exception {
+ if (shouldUseConcurrentPathForAdhoc(tableName)) {
+ return doCreateTask(taskType, tableName, taskName, taskConfigs);
+ }
+ synchronized (this) {
+ return doCreateTask(taskType, tableName, taskName, taskConfigs);
+ }
+ }
+
+ /// Resolves whether an ad-hoc [#createTask] may skip the controller-wide monitor. Every table the request
+ /// targets must opt in via [#resolveConcurrentScheduling] — the table-level override, else the cluster
+ /// default.
+ ///
+ /// Unlike [#shouldUseConcurrentPath], distributed locking is also required: without the monitor the ZK lock
+ /// is the only thing stopping two ad-hoc requests for the same table from generating tasks side by side.
+ ///
+ /// A table that cannot be resolved falls back to the legacy path, so an invalid name still fails inside
+ /// the synchronized call exactly as before.
+ protected boolean shouldUseConcurrentPathForAdhoc(String tableName) {
+ if (_distributedTaskLockManager == null) {
+ return false;
+ }
+ List<String> tableNameWithTypes;
+ try {
+ tableNameWithTypes = getTableNameWithTypes(tableName);
+ } catch (TableNotFoundException e) {
+ return false;
+ }
+ if (tableNameWithTypes.isEmpty()) {
+ return false;
+ }
+ for (String tableNameWithType : tableNameWithTypes) {
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig == null || !resolveConcurrentScheduling(tableConfig)) {
Review Comment:
The cron resolver skips a null config, but here it forces the legacy path. I think yours is the safer choice. Can you add a short comment so the difference reads as deliberate?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java:
##########
@@ -235,7 +236,56 @@ public void init() {
}
}
- public synchronized Map<String, String> createTask(String taskType, String tableName, @Nullable String taskName,
+ /// Creates ad-hoc tasks for the given table.
+ ///
+ /// Dispatches like [#scheduleTasks]: when every targeted table opts into concurrent scheduling (see
+ /// [#shouldUseConcurrentPathForAdhoc]) the call holds no controller-wide monitor, so ad-hoc generation for
+ /// one table no longer waits behind another's; otherwise it runs under `synchronized(this)` as before. Both
+ /// paths share [#doCreateTask].
+ public Map<String, String> createTask(String taskType, String tableName, @Nullable String taskName,
Review Comment:
With the monitor gone, a second ad-hoc call for the same table now hits the ZK lock and gets a 500 instead of waiting its turn. Is fail-fast what we want for callers that retry, and can the compat section say so?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java:
##########
@@ -1314,13 +1364,17 @@ public synchronized void reportMetrics(String taskType) {
}
}
- protected synchronized void addTaskTypeMetricsUpdaterIfNeeded(String taskType) {
- if (!_taskTypeMetricsUpdaterMap.containsKey(taskType)) {
- TaskTypeMetricsUpdater taskTypeMetricsUpdater = new TaskTypeMetricsUpdater(taskType, this);
+ /// Registers the per-task-type metrics listener once. Not `synchronized` on the manager: this runs from
Review Comment:
This comment tells the history of the bug, which fits the commit message better. Could it just say what the method guarantees (one listener per task type, no manager monitor)?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java:
##########
@@ -1314,13 +1364,17 @@ public synchronized void reportMetrics(String taskType) {
}
}
- protected synchronized void addTaskTypeMetricsUpdaterIfNeeded(String taskType) {
- if (!_taskTypeMetricsUpdaterMap.containsKey(taskType)) {
- TaskTypeMetricsUpdater taskTypeMetricsUpdater = new TaskTypeMetricsUpdater(taskType, this);
+ /// Registers the per-task-type metrics listener once. Not `synchronized` on the manager: this runs from
+ /// [#prepTaskQueue] on every generation path, the concurrent ones included, so taking the manager's monitor
+ /// here made each concurrent caller wait out any legacy-path generation still holding it. `computeIfAbsent`
+ /// keeps the once-per-type guarantee without that coupling.
+ protected void addTaskTypeMetricsUpdaterIfNeeded(String taskType) {
+ _taskTypeMetricsUpdaterMap.computeIfAbsent(taskType, type -> {
Review Comment:
This puts a ZK round trip (`subscribeDataChanges`) inside the CHM bin lock. It's fine for a once-per-type call, but can you walk me through why you chose it over subscribe-then-`putIfAbsent`?
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerConcurrentSchedulingTest.java:
##########
@@ -220,4 +233,244 @@ public void testShouldUseConcurrentPathAllTablesScopeHonorsPerTableOptOut()
assertFalse(manager.shouldUseConcurrentPath(new TaskSchedulingContext()));
}
+
+ // ---------- shouldUseConcurrentPathForAdhoc ----------
+
+ private PinotTaskManager adhocManager(boolean clusterDefault, boolean distributedLocking,
+ PinotHelixResourceManager resourceManager)
+ throws Exception {
+ PinotTaskManager manager = newManager(clusterDefault, resourceManager);
+ FieldUtils.writeField(manager, "_distributedTaskLockManager",
+ distributedLocking ? Mockito.mock(DistributedTaskLockManager.class) : null, true);
+ return manager;
+ }
+
+ @Test
+ public void testAdhocUsesConcurrentPathWhenTableOptsIn()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig(TABLE_A)).thenReturn(tableConfigWith(TABLE_A, Boolean.TRUE));
+ PinotTaskManager manager = adhocManager(false, true, rm);
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+
+ assertTrue(manager.shouldUseConcurrentPathForAdhoc(TABLE_A));
+ }
+
+ @Test
+ public void testAdhocInheritsClusterDefault()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig(TABLE_A)).thenReturn(tableConfigWith(TABLE_A, null));
+ PinotTaskManager manager = adhocManager(true, true, rm);
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+
+ assertTrue(manager.shouldUseConcurrentPathForAdhoc(TABLE_A));
+ }
+
+ @Test
+ public void testAdhocTableOptOutWinsOverClusterDefault()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig(TABLE_A)).thenReturn(tableConfigWith(TABLE_A, Boolean.FALSE));
+ PinotTaskManager manager = adhocManager(true, true, rm);
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+
+ assertFalse(manager.shouldUseConcurrentPathForAdhoc(TABLE_A));
+ }
+
+ /// Without distributed locking the monitor is the only same-table exclusion, so it must stay.
+ @Test
+ public void testAdhocStaysLegacyWithoutDistributedLocking()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig(TABLE_A)).thenReturn(tableConfigWith(TABLE_A, Boolean.TRUE));
+ PinotTaskManager manager = adhocManager(true, false, rm);
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+
+ assertFalse(manager.shouldUseConcurrentPathForAdhoc(TABLE_A));
+ }
+
+ /// A hybrid table name expands to both types; one opt-out keeps the whole request legacy.
+ @Test
+ public void testAdhocHybridTableNeedsEveryTypeToOptIn()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig("tableA_OFFLINE")).thenReturn(tableConfigWith("tableA_OFFLINE", Boolean.TRUE));
+ Mockito.when(rm.getTableConfig("tableA_REALTIME")).thenReturn(tableConfigWith("tableA_REALTIME", Boolean.FALSE));
+ PinotTaskManager manager = adhocManager(false, true, rm);
+ Mockito.doReturn(List.of("tableA_OFFLINE", "tableA_REALTIME")).when(manager).getTableNameWithTypes("tableA");
+
+ assertFalse(manager.shouldUseConcurrentPathForAdhoc("tableA"));
+ }
+
+ /// An unresolvable table must take the legacy path, where it fails exactly as it did before.
+ @Test
+ public void testAdhocUnknownTableFallsBackToLegacy()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ PinotTaskManager manager = adhocManager(true, true, rm);
+ Mockito.doThrow(new TableNotFoundException("missing")).when(manager).getTableNameWithTypes("missing");
+
+ assertFalse(manager.shouldUseConcurrentPathForAdhoc("missing"));
+ }
+
+ @Test
+ public void testAdhocMissingTableConfigFallsBackToLegacy()
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ PinotTaskManager manager = adhocManager(true, true, rm);
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+
+ assertFalse(manager.shouldUseConcurrentPathForAdhoc(TABLE_A));
+ }
+
+ // ---------- createTask: does another table's request wait on the monitor? ----------
+
+ /// Table A's generation parks inside createTask until released; table B's records that it got in.
+ /// [#prepTaskQueue] runs for real — only the Helix and ZK boundaries are mocked — because it is where a
+ /// concurrent caller can still meet the manager's monitor.
+ private PinotTaskManager blockingAdhocManager(boolean aOptsIn, boolean bOptsIn, CountDownLatch aEntered,
+ CountDownLatch releaseA, CountDownLatch bEntered)
+ throws Exception {
+ PinotHelixResourceManager rm = Mockito.mock(PinotHelixResourceManager.class);
+ Mockito.when(rm.getTableConfig(TABLE_A)).thenReturn(tableConfigWith(TABLE_A, aOptsIn));
+ Mockito.when(rm.getTableConfig(TABLE_B)).thenReturn(tableConfigWith(TABLE_B, bOptsIn));
+ Mockito.when(rm.getPropertyStore()).thenReturn(Mockito.mock(ZkHelixPropertyStore.class));
+ PinotTaskManager manager = adhocManager(false, true, rm);
+ FieldUtils.writeField(manager, "_helixTaskResourceManager", Mockito.mock(PinotHelixTaskResourceManager.class),
+ true);
+ FieldUtils.writeField(manager, "_taskTypeMetricsUpdaterMap", new ConcurrentHashMap<>(), true);
+
+ PinotTaskGenerator generator = Mockito.mock(PinotTaskGenerator.class);
+ Mockito.when(generator.generateTasks(any(TableConfig.class), anyMap())).thenAnswer(invocation -> {
+ TableConfig tableConfig = invocation.getArgument(0);
+ if (TABLE_A.equals(tableConfig.getTableName())) {
+ aEntered.countDown();
+ releaseA.await(30, TimeUnit.SECONDS);
+ } else {
+ bEntered.countDown();
+ }
+ return List.of();
+ });
+ Mockito.doReturn(true).when(manager).isTaskSchedulable(anyString(), anyList());
+ Mockito.doReturn("parent").when(manager).getParentTaskName(anyString(), anyString(), any());
+ Mockito.doReturn(List.of(TABLE_A)).when(manager).getTableNameWithTypes(TABLE_A);
+ Mockito.doReturn(List.of(TABLE_B)).when(manager).getTableNameWithTypes(TABLE_B);
+ Mockito.doReturn(generator).when(manager).getTaskGenerator(anyString(), anyString());
+ Mockito.doReturn(false).when(manager).isExceedingResourceUtilizationLimits(anyString());
+ Mockito.doReturn(null).when(manager).acquireTaskLock(anyString(), anyString(), anyString());
Review Comment:
All the behaviour tests stub `acquireTaskLock` to null, so same-table exclusion, which is the safety argument for dropping the monitor, isn't tested. Could you add a two-calls-same-table test with a real or fake lock manager?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Honor concurrent scheduling in ad-hoc createTask [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19660:
URL: https://github.com/apache/pinot/pull/19660#issuecomment-5815818577
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19660?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 0.00%. Comparing base ([`39ef02c`](https://app.codecov.io/gh/apache/pinot/commit/39ef02cc5f272e59e30432c52cbcc08285d1676c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a269f32`](https://app.codecov.io/gh/apache/pinot/commit/a269f32908c69bc98a74e13604f006005eb54564?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
> :exclamation: There is a different number of reports uploaded between BASE (39ef02c) and HEAD (a269f32). Click for more details.
>
> <details><summary>HEAD has 18 uploads less than BASE</summary>
>
>| Flag | BASE (39ef02c) | HEAD (a269f32) |
>|------|------|------|
>|lane-a|2|1|
>|integration|4|2|
>|java-25|6|2|
>|temurin|6|2|
>|lane-b|2|1|
>|integration1|2|0|
>|unittests1|1|0|
>|unittests|2|0|
>|unittests2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19660 +/- ##
=============================================
- Coverage 67.95% 0.00% -67.96%
=============================================
Files 3511 3 -3508
Lines 227477 6 -227471
Branches 35994 0 -35994
=============================================
- Hits 154573 0 -154573
+ Misses 60755 6 -60749
+ Partials 12149 0 -12149
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-100.00%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-67.96%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-100.00%)` | :arrow_down: |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-67.96%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19660/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19660?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org