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: [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]
