You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "somandal (via GitHub)" <gi...@apache.org> on 2025/08/15 15:08:40 UTC
[PR] Change Periodic task implementations that use synchronized methods to use a separate lock instead [pinot]
somandal opened a new pull request, #16613:
URL: https://github.com/apache/pinot/pull/16613
Today the `BasePeriodicTask` has it's `stop()` and `start()` methods marked as `synchronized`. Some of the periodic tasks call `synchronized` methods under `processTables`/`processTable` calls. The main purpose behind this task level synchronization is to prevent scheduled tasks and tasks that are submitted ad-hoc from contending with each other. For such tasks, during controller shutdown we've seen scenarios where if a controller periodic task is long running, the synchronized `stop()` gets blocked on getting the lock, delaying the controller shutdown phase.
To address this, this PR adds a separate `_taskLock` object which can be used in methods like `processTables`/`processTable` to serve the purpose of ensuring that multiple tasks don't run at the same time. Using a separate lock prevents the long wait on shutdown to get the synchronized `stop()` lock.
cc @Jackie-Jiang @xiangfu0
--
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] Change BasePeriodicTask to use a lock object for start / stop rather than synchronizing the methods [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal merged PR #16613:
URL: https://github.com/apache/pinot/pull/16613
--
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] Change Periodic task implementations that use synchronized methods to use a separate lock instead [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16613:
URL: https://github.com/apache/pinot/pull/16613#discussion_r2279873670
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -67,19 +67,19 @@ public RebalanceChecker(TableRebalanceManager tableRebalanceManager,
@Override
protected void processTables(List<String> tableNamesWithType, Properties periodicTaskProperties) {
- int numTables = tableNamesWithType.size();
- LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
- int numTablesProcessed = retryRebalanceTables(new HashSet<>(tableNamesWithType));
- _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED, _taskName,
- numTablesProcessed);
- LOGGER.info("Finish processing {}/{} tables in task: {}", numTablesProcessed, numTables, _taskName);
+ synchronized (_taskLock) {
+ int numTables = tableNamesWithType.size();
+ LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
+ // Rare but the task may be executed by more than one threads because user can trigger the periodic task to run
Review Comment:
The comment contains a grammatical error. 'more than one threads' should be 'more than one thread'.
```suggestion
// Rare but the task may be executed by more than one thread because user can trigger the periodic task to run
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java:
##########
@@ -619,102 +637,111 @@ public synchronized TaskSchedulingInfo scheduleTaskForDatabase(String taskType,
* - list of task scheduling errors if any
*/
@Deprecated(forRemoval = true)
- public synchronized TaskSchedulingInfo scheduleTaskForTable(String taskType, String tableNameWithType,
+ public TaskSchedulingInfo scheduleTaskForTable(String taskType, String tableNameWithType,
@Nullable String minionInstanceTag) {
- TaskSchedulingContext context = new TaskSchedulingContext()
- .setTasksToSchedule(Collections.singleton(taskType))
- .setTablesToSchedule(Collections.singleton(tableNameWithType))
- .setMinionInstanceTag(minionInstanceTag);
- return scheduleTasks(context).get(taskType);
+ synchronized (_taskLock) {
+ TaskSchedulingContext context = new TaskSchedulingContext()
+ .setTasksToSchedule(Collections.singleton(taskType))
+ .setTablesToSchedule(Collections.singleton(tableNameWithType))
+ .setMinionInstanceTag(minionInstanceTag);
+ return scheduleTasks(context).get(taskType);
+ }
}
/**
* Helper method to schedule tasks (all task types) for the given tables that have the tasks enabled.
* Returns a map from the task type to the {@link TaskSchedulingInfo} of the tasks scheduled.
*/
@Deprecated(forRemoval = true)
- protected synchronized Map<String, TaskSchedulingInfo> scheduleTasks(List<String> tableNamesWithType,
+ protected Map<String, TaskSchedulingInfo> scheduleTasks(List<String> tableNamesWithType,
boolean isLeader, @Nullable String minionInstanceTag) {
- TaskSchedulingContext context = new TaskSchedulingContext()
- .setTablesToSchedule(new HashSet<>(tableNamesWithType))
- .setLeader(isLeader)
- .setMinionInstanceTag(minionInstanceTag);
- return scheduleTasks(context);
+ synchronized (_taskLock) {
+ TaskSchedulingContext context = new TaskSchedulingContext()
+ .setTablesToSchedule(new HashSet<>(tableNamesWithType))
+ .setLeader(isLeader)
+ .setMinionInstanceTag(minionInstanceTag);
+ return scheduleTasks(context);
+ }
}
/**
* Helper method to schedule tasks (all task types) for the given tables that have the tasks enabled.
* Returns a map from the task type to the {@link TaskSchedulingInfo} of the tasks scheduled.
*/
- public synchronized Map<String, TaskSchedulingInfo> scheduleTasks(TaskSchedulingContext context) {
- _controllerMetrics.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED, 1L);
-
- Map<String, List<TableConfig>> enabledTableConfigMap = new HashMap<>();
- Set<String> targetTables = context.getTablesToSchedule();
- Set<String> targetDatabases = context.getDatabasesToSchedule();
- Set<String> tasksToSchedule = context.getTasksToSchedule();
- Set<String> consolidatedTables = new HashSet<>();
- if (targetTables != null) {
- consolidatedTables.addAll(targetTables);
- }
- if (targetDatabases != null) {
- targetDatabases.forEach(database ->
- consolidatedTables.addAll(_pinotHelixResourceManager.getAllTables(database)));
- }
- for (String tableNameWithType : consolidatedTables.isEmpty()
- ? _pinotHelixResourceManager.getAllTables() : consolidatedTables) {
- TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
- if (tableConfig != null && tableConfig.getTaskConfig() != null) {
- Set<String> enabledTaskTypes = tableConfig.getTaskConfig().getTaskTypeConfigsMap().keySet();
- Set<String> validTasks;
- if (tasksToSchedule == null || tasksToSchedule.isEmpty()) {
- // if no specific task types are provided schedule for all tasks
- validTasks = enabledTaskTypes;
- } else {
- validTasks = new HashSet<>(tasksToSchedule);
- validTasks.retainAll(enabledTaskTypes);
+ public Map<String, TaskSchedulingInfo> scheduleTasks(TaskSchedulingContext context) {
+ synchronized (_taskLock) {
+ _controllerMetrics.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED, 1L);
+
+ Map<String, List<TableConfig>> enabledTableConfigMap = new HashMap<>();
+ Set<String> targetTables = context.getTablesToSchedule();
+ Set<String> targetDatabases = context.getDatabasesToSchedule();
+ Set<String> tasksToSchedule = context.getTasksToSchedule();
+ Set<String> consolidatedTables = new HashSet<>();
+ if (targetTables != null) {
+ consolidatedTables.addAll(targetTables);
+ }
+ if (targetDatabases != null) {
+ targetDatabases.forEach(database ->
+ consolidatedTables.addAll(_pinotHelixResourceManager.getAllTables(database)));
+ }
+ for (String tableNameWithType : consolidatedTables.isEmpty()
+ ? _pinotHelixResourceManager.getAllTables() : consolidatedTables) {
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig != null && tableConfig.getTaskConfig() != null) {
+ Set<String> enabledTaskTypes = tableConfig.getTaskConfig().getTaskTypeConfigsMap().keySet();
+ Set<String> validTasks;
+ if (tasksToSchedule == null || tasksToSchedule.isEmpty()) {
+ // if no specific task types are provided schedule for all tasks
+ validTasks = enabledTaskTypes;
+ } else {
+ validTasks = new HashSet<>(tasksToSchedule);
+ validTasks.retainAll(enabledTaskTypes);
+ }
+ for (String taskType : validTasks) {
+ enabledTableConfigMap.computeIfAbsent(taskType, k -> new ArrayList<>()).add(tableConfig);
+ }
}
- for (String taskType : validTasks) {
- enabledTableConfigMap.computeIfAbsent(taskType, k -> new ArrayList<>()).add(tableConfig);
+ }
+
+ // Generate each type of tasks
+ Map<String, TaskSchedulingInfo> tasksScheduled = new HashMap<>();
+ for (Map.Entry<String, List<TableConfig>> entry : enabledTableConfigMap.entrySet()) {
+ String taskType = entry.getKey();
+ List<TableConfig> enabledTableConfigs = entry.getValue();
+ PinotTaskGenerator taskGenerator = _taskGeneratorRegistry.getTaskGenerator(taskType);
+ if (taskGenerator != null) {
+ _helixTaskResourceManager.ensureTaskQueueExists(taskType);
+ addTaskTypeMetricsUpdaterIfNeeded(taskType);
+ tasksScheduled.put(taskType, scheduleTask(taskGenerator, enabledTableConfigs, context.isLeader(),
+ context.getMinionInstanceTag(), context.getTriggeredBy()));
+ } else {
+ List<String> enabledTables =
+ enabledTableConfigs.stream().map(TableConfig::getTableName).collect(Collectors.toList());
+ String message =
Review Comment:
This line appears to be incorrectly indented with extra spaces. The assignment should align with the surrounding code block.
--
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] Change Periodic task implementations that use synchronized methods to use a separate lock instead [pinot]
Posted by "somandal (via GitHub)" <gi...@apache.org>.
somandal commented on PR #16613:
URL: https://github.com/apache/pinot/pull/16613#issuecomment-3192960020
> It might be easier if we introduce a life-cycle lock within the `BasePeriodicTask` to only synchronize the life-cycle changes. This way all life-cycle changes won't be blocked
done, please take a look. thanks for the suggestion!
--
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