diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java index 24cd444dc592..32efb5ebc20b 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTaskRestletResource.java @@ -80,6 +80,8 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingInfo; import org.apache.pinot.controller.util.CompletionServiceHelper; import org.apache.pinot.core.auth.Actions; import org.apache.pinot.core.auth.Authorize; @@ -646,21 +648,36 @@ public Map scheduleTasks( Map response = new HashMap<>(); List generationErrors = new ArrayList<>(); List schedulingErrors = new ArrayList<>(); + TaskSchedulingContext context = new TaskSchedulingContext() + .setTriggeredBy(PinotTaskManager.Triggers.MANUAL_TRIGGER.name()) + .setMinionInstanceTag(minionInstanceTag) + .setLeader(false); if (taskType != null) { + Map> tableToTaskNamesMap = new HashMap<>(); + Set taskTypes = new HashSet<>(1); + taskTypes.add(taskType); // Schedule task for the given task type - PinotTaskManager.TaskSchedulingInfo taskInfos = tableName != null - ? _pinotTaskManager.scheduleTaskForTable(taskType, DatabaseUtils.translateTableName(tableName, headers), - minionInstanceTag) - : _pinotTaskManager.scheduleTaskForDatabase(taskType, database, minionInstanceTag); + if (tableName != null) { + tableToTaskNamesMap.put(DatabaseUtils.translateTableName(tableName, headers), taskTypes); + } else { + _pinotHelixResourceManager.getAllTables(database).forEach(table -> tableToTaskNamesMap.put(table, taskTypes)); + } + context.setTableToTaskNamesMap(tableToTaskNamesMap); + TaskSchedulingInfo taskInfos = _pinotTaskManager.scheduleTasks(context).get(taskType); response.put(taskType, StringUtils.join(taskInfos.getScheduledTaskNames(), ',')); generationErrors.addAll(taskInfos.getGenerationErrors()); schedulingErrors.addAll(taskInfos.getSchedulingErrors()); } else { + Map> tableToTaskNamesMap = new HashMap<>(); // Schedule tasks for all task types - Map allTaskInfos = tableName != null - ? _pinotTaskManager.scheduleAllTasksForTable(DatabaseUtils.translateTableName(tableName, headers), - minionInstanceTag) - : _pinotTaskManager.scheduleAllTasksForDatabase(database, minionInstanceTag); + if (tableName != null) { + tableToTaskNamesMap.put(DatabaseUtils.translateTableName(tableName, headers), null); + } else { + _pinotHelixResourceManager.getAllTables(database) + .forEach(table -> tableToTaskNamesMap.put(table, null)); + } + context.setTableToTaskNamesMap(tableToTaskNamesMap); + Map allTaskInfos = _pinotTaskManager.scheduleTasks(context); allTaskInfos.forEach((key, value) -> { if (value.getScheduledTaskNames() != null) { response.put(key, String.join(",", value.getScheduledTaskNames())); diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/CronJobScheduleJob.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/CronJobScheduleJob.java index f9b250b2bcd4..39c06644a32e 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/CronJobScheduleJob.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/CronJobScheduleJob.java @@ -64,8 +64,10 @@ public void execute(JobExecutionContext jobExecutionContext) ControllerMeter.CRON_SCHEDULER_JOB_SKIPPED, 1L); return; } + TaskSchedulingContext context = new TaskSchedulingContext(table, taskType) + .setTriggeredBy(PinotTaskManager.Triggers.CRON_TRIGGER.name()); long jobStartTime = System.currentTimeMillis(); - pinotTaskManager.scheduleTaskForTable(taskType, table, null); + pinotTaskManager.scheduleTasks(context); LOGGER.info("Finished CronJob: table - {}, task - {}, next runtime is {}", table, taskType, jobExecutionContext.getNextFireTime()); pinotTaskManager.getControllerMetrics().addTimedTableValue(PinotTaskManager.getCronJobName(table, taskType), diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java index 87580c30b045..bbbb3fcee26f 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java @@ -876,6 +876,11 @@ private synchronized TaskDebugInfo getTaskDebugInfo(WorkflowContext workflowCont if (jobFinishTimeMs > 0) { taskDebugInfo.setFinishTime(DateTimeUtils.epochToDefaultDateFormat(jobFinishTimeMs)); } + String triggeredBy = jobConfig.getTaskConfigMap().values().stream().findFirst() + .map(TaskConfig::getConfigMap) + .map(taskConfigs -> taskConfigs.get(PinotTaskManager.TRIGGERED_BY)) + .orElse(""); + taskDebugInfo.setTriggeredBy(triggeredBy); Set partitionSet = jobContext.getPartitionSet(); TaskCount subtaskCount = new TaskCount(); for (int partition : partitionSet) { @@ -890,6 +895,7 @@ private synchronized TaskDebugInfo getTaskDebugInfo(WorkflowContext workflowCont String taskIdForPartition = jobContext.getTaskIdForPartition(partition); subtaskDebugInfo.setTaskId(taskIdForPartition); subtaskDebugInfo.setState(partitionState); + subtaskDebugInfo.setTriggeredBy(triggeredBy); long subtaskStartTimeMs = jobContext.getPartitionStartTime(partition); if (subtaskStartTimeMs > 0) { subtaskDebugInfo.setStartTime(DateTimeUtils.epochToDefaultDateFormat(subtaskStartTimeMs)); @@ -987,7 +993,8 @@ public Map> getTaskMetadataLastUpdateTimeMs() { return MinionTaskMetadataUtils.getAllTaskMetadataLastUpdateTimeMs(propertyStore); } - @JsonPropertyOrder({"taskState", "subtaskCount", "startTime", "executionStartTime", "finishTime", "subtaskInfos"}) + @JsonPropertyOrder({"taskState", "subtaskCount", "startTime", "executionStartTime", "finishTime", "triggeredBy", + "subtaskInfos"}) @JsonInclude(JsonInclude.Include.NON_NULL) public static class TaskDebugInfo { // Time at which the task (which may have multiple subtasks) got created. @@ -998,6 +1005,7 @@ public static class TaskDebugInfo { private String _finishTime; private TaskState _taskState; private TaskCount _subtaskCount; + private String _triggeredBy; private List _subtaskInfos; public TaskDebugInfo() { @@ -1046,6 +1054,15 @@ public TaskState getTaskState() { return _taskState; } + public String getTriggeredBy() { + return _triggeredBy; + } + + public TaskDebugInfo setTriggeredBy(String triggeredBy) { + _triggeredBy = triggeredBy; + return this; + } + public TaskCount getSubtaskCount() { return _subtaskCount; } @@ -1055,7 +1072,7 @@ public List getSubtaskInfos() { } } - @JsonPropertyOrder({"taskId", "state", "startTime", "finishTime", "participant", "info", "taskConfig"}) + @JsonPropertyOrder({"taskId", "state", "startTime", "finishTime", "participant", "info", "triggeredBy", "taskConfig"}) @JsonInclude(JsonInclude.Include.NON_NULL) public static class SubtaskDebugInfo { private String _taskId; @@ -1064,6 +1081,7 @@ public static class SubtaskDebugInfo { private String _finishTime; private String _participant; private String _info; + private String _triggeredBy; private PinotTaskConfig _taskConfig; public SubtaskDebugInfo() { @@ -1121,6 +1139,15 @@ public String getInfo() { return _info; } + public String getTriggeredBy() { + return _triggeredBy; + } + + public SubtaskDebugInfo setTriggeredBy(String triggeredBy) { + _triggeredBy = triggeredBy; + return this; + } + public PinotTaskConfig getTaskConfig() { return _taskConfig; } diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java index 93002f9100d8..e3e477ced2ee 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java @@ -18,7 +18,6 @@ */ package org.apache.pinot.controller.helix.core.minion; -import com.google.common.base.Preconditions; import java.io.PrintWriter; import java.io.StringWriter; import java.util.ArrayList; @@ -94,6 +93,7 @@ public class PinotTaskManager extends ControllerPeriodicTask { private static final String TABLE_CONFIG_PARENT_PATH = "/CONFIGS/TABLE"; private static final String TABLE_CONFIG_PATH_PREFIX = "/CONFIGS/TABLE/"; private static final String TASK_QUEUE_PATH_PATTERN = "/TaskRebalancer/TaskQueue_%s/Context"; + public static final String TRIGGERED_BY = "triggeredBy"; private final PinotHelixTaskResourceManager _helixTaskResourceManager; private final ClusterInfoAccessor _clusterInfoAccessor; @@ -114,6 +114,10 @@ public class PinotTaskManager extends ControllerPeriodicTask { private final TaskManagerStatusCache _taskManagerStatusCache; + public enum Triggers { + CRON_TRIGGER, MANUAL_TRIGGER, ADHOC_TRIGGER + } + public PinotTaskManager(PinotHelixTaskResourceManager helixTaskResourceManager, PinotHelixResourceManager helixResourceManager, LeadControllerManager leadControllerManager, ControllerConf controllerConf, ControllerMetrics controllerMetrics, @@ -208,6 +212,8 @@ public Map createTask(String taskType, String tableName, @Nullab LOGGER.warn("No ad-hoc task generated for task type: {}", taskType); continue; } + pinotTaskConfigs.forEach(pinotTaskConfig -> + pinotTaskConfig.getConfigs().computeIfAbsent(TRIGGERED_BY, k -> Triggers.ADHOC_TRIGGER.name())); LOGGER.info("Submitting ad-hoc task for task type: {} with task configs: {}", taskType, pinotTaskConfigs); _controllerMetrics.addMeteredTableValue(taskType, ControllerMeter.NUMBER_ADHOC_TASKS_SUBMITTED, 1); responseMap.put(tableNameWithType, @@ -483,92 +489,34 @@ public void registerTaskGenerator(PinotTaskGenerator taskGenerator) { _taskGeneratorRegistry.registerTaskGenerator(taskGenerator); } - /** - * Schedules tasks (all task types) for all tables. - * It might be called from the non-leader controller. - * Returns a map from the task type to the {@link TaskSchedulingInfo} of tasks scheduled. - */ - public synchronized Map scheduleAllTasksForAllTables(@Nullable String minionInstanceTag) { - return scheduleTasks(_pinotHelixResourceManager.getAllTables(), false, minionInstanceTag); - } - - /** - * Schedules tasks (all task types) for all tables in the given database. - * It might be called from the non-leader controller. - * Returns a map from the task type to the {@link TaskSchedulingInfo} of tasks scheduled. - */ - public synchronized Map scheduleAllTasksForDatabase(@Nullable String database, - @Nullable String minionInstanceTag) { - return scheduleTasks(_pinotHelixResourceManager.getAllTables(database), false, minionInstanceTag); - } - - /** - * Schedules tasks (all task types) for the given table. - * It might be called from the non-leader controller. - * Returns a map from the task type to the {@link TaskSchedulingInfo} of tasks scheduled. - */ - public synchronized Map scheduleAllTasksForTable(String tableNameWithType, - @Nullable String minionInstanceTag) { - return scheduleTasks(List.of(tableNameWithType), false, minionInstanceTag); - } - - /** - * Schedules task for the given task type for all tables. - * It might be called from the non-leader controller. - * Returns {@link TaskSchedulingInfo} which consists - * - list of scheduled task names (empty list if nothing to schedule), - * or {@code null} if no task is scheduled due to scheduling errors. - * - list of task generation errors if any - * - list of task scheduling errors if any - */ - public synchronized TaskSchedulingInfo scheduleTaskForAllTables(String taskType, @Nullable String minionInstanceTag) { - return scheduleTask(taskType, _pinotHelixResourceManager.getAllTables(), minionInstanceTag); - } - - /** - * Schedules task for the given task type for all tables in the given database. - * It might be called from the non-leader controller. - * Returns {@link TaskSchedulingInfo} which consists - * - list of scheduled task names (empty list if nothing to schedule), - * or {@code null} if no task is scheduled due to scheduling errors. - * - list of task generation errors if any - * - list of task scheduling errors if any - */ - public synchronized TaskSchedulingInfo scheduleTaskForDatabase(String taskType, @Nullable String database, - @Nullable String minionInstanceTag) { - return scheduleTask(taskType, _pinotHelixResourceManager.getAllTables(database), minionInstanceTag); - } - - /** - * Schedules task for the given task type for the give table. - * It might be called from the non-leader controller. - * Returns {@link TaskSchedulingInfo} which consists - * - list of scheduled task names (empty list if nothing to schedule), - * or {@code null} if no task is scheduled due to scheduling errors. - * - list of task generation errors if any - * - list of task scheduling errors if any - */ - public synchronized TaskSchedulingInfo scheduleTaskForTable(String taskType, String tableNameWithType, - @Nullable String minionInstanceTag) { - return scheduleTask(taskType, List.of(tableNameWithType), minionInstanceTag); - } - /** * 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. */ - protected synchronized Map scheduleTasks(List tableNamesWithType, - boolean isLeader, @Nullable String minionInstanceTag) { + public synchronized Map scheduleTasks(TaskSchedulingContext context) { _controllerMetrics.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED, 1L); - // Scan all table configs to get the tables with tasks enabled Map> enabledTableConfigMap = new HashMap<>(); - for (String tableNameWithType : tableNamesWithType) { + Map> tableToTasksMap = context.getTableToTaskNamesMap(); + if (context.getTableToTaskNamesMap().isEmpty()) { + _pinotHelixResourceManager.getAllTables().forEach(table -> tableToTasksMap.put(table, null)); + } + for (Map.Entry> entry : context.getTableToTaskNamesMap().entrySet()) { + String tableNameWithType = entry.getKey(); + Set taskNames = entry.getValue(); TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType); if (tableConfig != null && tableConfig.getTaskConfig() != null) { Set enabledTaskTypes = tableConfig.getTaskConfig().getTaskTypeConfigsMap().keySet(); - for (String enabledTaskType : enabledTaskTypes) { - enabledTableConfigMap.computeIfAbsent(enabledTaskType, k -> new ArrayList<>()).add(tableConfig); + Set validTasks; + if (taskNames == null || taskNames.isEmpty()) { + // if no specific task types are provided schedule for all tasks + validTasks = enabledTaskTypes; + } else { + validTasks = new HashSet<>(taskNames); + validTasks.retainAll(enabledTaskTypes); + } + for (String taskType : validTasks) { + enabledTableConfigMap.computeIfAbsent(taskType, k -> new ArrayList<>()).add(tableConfig); } } } @@ -579,13 +527,14 @@ protected synchronized Map scheduleTasks(List enabledTableConfigs = entry.getValue(); PinotTaskGenerator taskGenerator = _taskGeneratorRegistry.getTaskGenerator(taskType); - List enabledTables = - enabledTableConfigs.stream().map(TableConfig::getTableName).collect(Collectors.toList()); if (taskGenerator != null) { _helixTaskResourceManager.ensureTaskQueueExists(taskType); addTaskTypeMetricsUpdaterIfNeeded(taskType); - tasksScheduled.put(taskType, scheduleTask(taskGenerator, enabledTableConfigs, isLeader, minionInstanceTag)); + tasksScheduled.put(taskType, scheduleTask(taskGenerator, enabledTableConfigs, context.isLeader(), + context.getMinionInstanceTag(), context.getTriggeredBy())); } else { + List enabledTables = + enabledTableConfigs.stream().map(TableConfig::getTableName).collect(Collectors.toList()); String message = "Task type: " + taskType + " is not registered, cannot enable it for tables: " + enabledTables; LOGGER.warn(message); TaskSchedulingInfo taskSchedulingInfo = new TaskSchedulingInfo(); @@ -597,26 +546,6 @@ protected synchronized Map scheduleTasks(List tables, - @Nullable String minionInstanceTag) { - PinotTaskGenerator taskGenerator = _taskGeneratorRegistry.getTaskGenerator(taskType); - Preconditions.checkState(taskGenerator != null, "Task type: %s is not registered", taskType); - - // Scan all table configs to get the tables with task enabled - List enabledTableConfigs = new ArrayList<>(); - for (String tableNameWithType : tables) { - TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType); - if (tableConfig != null && tableConfig.getTaskConfig() != null && tableConfig.getTaskConfig() - .isTaskTypeEnabled(taskType)) { - enabledTableConfigs.add(tableConfig); - } - } - - _helixTaskResourceManager.ensureTaskQueueExists(taskType); - addTaskTypeMetricsUpdaterIfNeeded(taskType); - return scheduleTask(taskGenerator, enabledTableConfigs, false, minionInstanceTag); - } - /** * Helper method to schedule task with the given task generator for the given tables that have the task enabled. * Returns @@ -626,8 +555,8 @@ protected synchronized TaskSchedulingInfo scheduleTask(String taskType, List enabledTableConfigs, - boolean isLeader, @Nullable String minionInstanceTagForTask) { - TaskSchedulingInfo response = new TaskSchedulingInfo(); + boolean isLeader, @Nullable String minionInstanceTagForTask, String triggeredBy) { + TaskSchedulingInfo response = new TaskSchedulingInfo(); String taskType = taskGenerator.getTaskType(); List enabledTables = enabledTableConfigs.stream().map(TableConfig::getTableName).collect(Collectors.toList()); @@ -693,6 +622,8 @@ protected TaskSchedulingInfo scheduleTask(PinotTaskGenerator taskGenerator, List // This might lead to lot of logs, maybe sum it up and move outside the loop LOGGER.info("Submitting {} tasks for task type: {} to minionInstance: {} with task configs: {}", numTasks, taskType, minionInstanceTag, pinotTaskConfigs); + pinotTaskConfigs.forEach(pinotTaskConfig -> + pinotTaskConfig.getConfigs().computeIfAbsent(TRIGGERED_BY, k -> triggeredBy)); String submittedTaskName = _helixTaskResourceManager.submitTask(pinotTaskConfigs, minionInstanceTag, taskGenerator.getTaskTimeoutMs(), taskGenerator.getNumConcurrentTasksPerInstance(), taskGenerator.getMaxAttemptsPerTask()); @@ -718,7 +649,11 @@ protected TaskSchedulingInfo scheduleTask(PinotTaskGenerator taskGenerator, List @Override protected void processTables(List tableNamesWithType, Properties taskProperties) { - scheduleTasks(tableNamesWithType, true, null); + TaskSchedulingContext context = new TaskSchedulingContext(tableNamesWithType) + .setLeader(true) + .setTriggeredBy(Triggers.CRON_TRIGGER.name()); + // cron schedule + scheduleTasks(context); } @Override @@ -781,36 +716,4 @@ protected boolean isTaskSchedulable(String taskType, List tables) { } return true; } - - public static class TaskSchedulingInfo { - private List _scheduledTaskNames; - private final List _generationErrors = new ArrayList<>(); - private final List _schedulingErrors = new ArrayList<>(); - - @Nullable - public List getScheduledTaskNames() { - return _scheduledTaskNames; - } - - public TaskSchedulingInfo setScheduledTaskNames(List scheduledTaskNames) { - _scheduledTaskNames = scheduledTaskNames; - return this; - } - - public List getGenerationErrors() { - return _generationErrors; - } - - public void addGenerationError(String generationError) { - _generationErrors.add(generationError); - } - - public List getSchedulingErrors() { - return _schedulingErrors; - } - - public void addSchedulingError(String schedulingError) { - _schedulingErrors.add(schedulingError); - } - } } diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingContext.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingContext.java new file mode 100644 index 000000000000..da524a267ac9 --- /dev/null +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingContext.java @@ -0,0 +1,74 @@ +package org.apache.pinot.controller.helix.core.minion; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + + +/** + * Wrapper class to manage all the inputs passed to schedule a task on minion + * _tableToTaskNamesMap is a map of table name and its consecutive task types for which task needs to be scheduled + *

+ * Few special cases to note : + *

  • If the value for a table name entry is null or empty set then + * it will end up scheduling all the configured tasks on that table. + *
  • If the _tableToTaskNamesMap is empty then it will end up scheduling + * all the configured tasks for all the tables. + */ +public class TaskSchedulingContext { + private Map> _tableToTaskNamesMap; + private String _triggeredBy; + private String _minionInstanceTag; + private boolean _isLeader; + + public TaskSchedulingContext() { + _tableToTaskNamesMap = new HashMap<>(); + } + + public TaskSchedulingContext(List tableNames) { + _tableToTaskNamesMap = new HashMap<>(tableNames.size()); + tableNames.forEach(tableName -> _tableToTaskNamesMap.put(tableName, null)); + } + + public TaskSchedulingContext(String tableName, String taskName) { + _tableToTaskNamesMap = new HashMap<>(1); + _tableToTaskNamesMap.put(tableName, Set.of(taskName)); + } + + public Map> getTableToTaskNamesMap() { + return _tableToTaskNamesMap; + } + + public TaskSchedulingContext setTableToTaskNamesMap(Map> tableToTaskNamesMap) { + _tableToTaskNamesMap = tableToTaskNamesMap; + return this; + } + + public String getTriggeredBy() { + return _triggeredBy; + } + + public TaskSchedulingContext setTriggeredBy(String triggeredBy) { + _triggeredBy = triggeredBy; + return this; + } + + public String getMinionInstanceTag() { + return _minionInstanceTag; + } + + public TaskSchedulingContext setMinionInstanceTag(String minionInstanceTag) { + _minionInstanceTag = minionInstanceTag; + return this; + } + + public boolean isLeader() { + return _isLeader; + } + + public TaskSchedulingContext setLeader(boolean leader) { + _isLeader = leader; + return this; + } +} diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingInfo.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingInfo.java new file mode 100644 index 000000000000..e55083884443 --- /dev/null +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskSchedulingInfo.java @@ -0,0 +1,38 @@ +package org.apache.pinot.controller.helix.core.minion; + +import java.util.ArrayList; +import java.util.List; +import javax.annotation.Nullable; + + +public class TaskSchedulingInfo { + private List _scheduledTaskNames; + private final List _generationErrors = new ArrayList<>(); + private final List _schedulingErrors = new ArrayList<>(); + + @Nullable + public List getScheduledTaskNames() { + return _scheduledTaskNames; + } + + public TaskSchedulingInfo setScheduledTaskNames(List scheduledTaskNames) { + _scheduledTaskNames = scheduledTaskNames; + return this; + } + + public List getGenerationErrors() { + return _generationErrors; + } + + public void addGenerationError(String generationError) { + _generationErrors.add(generationError); + } + + public List getSchedulingErrors() { + return _schedulingErrors; + } + + public void addSchedulingError(String schedulingError) { + _schedulingErrors.add(schedulingError); + } +} diff --git a/pinot-controller/src/main/resources/app/pages/SubTaskDetail.tsx b/pinot-controller/src/main/resources/app/pages/SubTaskDetail.tsx index f94cd4d82048..27ab5dc27331 100644 --- a/pinot-controller/src/main/resources/app/pages/SubTaskDetail.tsx +++ b/pinot-controller/src/main/resources/app/pages/SubTaskDetail.tsx @@ -113,6 +113,9 @@ const TaskDetail = (props) => { Finish Time: {get(taskDebugData, 'finishTime', '')} + + Triggered By: {get(taskDebugData, 'triggeredBy', '')} + Minion Host Name: {get(taskDebugData, 'participant', '')} diff --git a/pinot-controller/src/main/resources/app/pages/TaskDetail.tsx b/pinot-controller/src/main/resources/app/pages/TaskDetail.tsx index f825cfbd0570..eff82fd3fd25 100644 --- a/pinot-controller/src/main/resources/app/pages/TaskDetail.tsx +++ b/pinot-controller/src/main/resources/app/pages/TaskDetail.tsx @@ -124,6 +124,9 @@ const TaskDetail = (props) => { Finish Time: {get(taskDebugData, 'finishTime', '')} + + Triggered By: {get(taskDebugData, 'triggeredBy', '')} + Number of Sub Tasks: {get(taskDebugData, 'subtaskCount.total', '')} diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerStatelessTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerStatelessTest.java index 132e10979673..f0f2f6903cc3 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerStatelessTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerStatelessTest.java @@ -192,10 +192,13 @@ public List generateTasks(List tableConfigs) { public void testPinotTaskManagerScheduleTaskWithStoppedTaskQueue() throws Exception { testValidateTaskGeneration(taskManager -> { + String taskName = "SegmentGenerationAndPushTask"; + TaskSchedulingContext context = new TaskSchedulingContext("myTable", taskName); // Validate schedule tasks for table when task queue is in stopped state - List taskIDs = taskManager.scheduleTaskForTable("SegmentGenerationAndPushTask", "myTable", null) - .getScheduledTaskNames(); - assertNull(taskIDs); + TaskSchedulingInfo info = taskManager.scheduleTasks(context).get(taskName); + assertNotNull(info); + assertNull(info.getScheduledTaskNames()); + assertFalse(info.getSchedulingErrors().isEmpty()); return null; }); } diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/MinionTaskTestUtils.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/MinionTaskTestUtils.java index 849a8b8bfdb5..2291bbb85828 100644 --- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/MinionTaskTestUtils.java +++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/MinionTaskTestUtils.java @@ -20,6 +20,8 @@ import java.util.Map; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingInfo; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; @@ -29,25 +31,18 @@ public class MinionTaskTestUtils { private MinionTaskTestUtils() { } - public static void assertNoTaskSchedule(String tableNameWithType, String taskType, PinotTaskManager taskManager) { - PinotTaskManager.TaskSchedulingInfo info = - taskManager.scheduleAllTasksForTable(tableNameWithType, null).get(taskType); - assertNoTaskSchedule(info); - } - - public static void assertNoTaskSchedule(String taskType, PinotTaskManager taskManager) { - PinotTaskManager.TaskSchedulingInfo info = taskManager.scheduleTaskForAllTables(taskType, null); - assertNoTaskSchedule(info); - } - - public static void assertNoTaskSchedule(PinotTaskManager taskManager) { - Map infoMap = taskManager.scheduleAllTasksForAllTables(null); + public static void assertNoTaskSchedule(TaskSchedulingContext context, PinotTaskManager taskManager) { + Map infoMap = taskManager.scheduleTasks(context); infoMap.forEach((key, value) -> assertNoTaskSchedule(value)); } - public static void assertNoTaskSchedule(PinotTaskManager.TaskSchedulingInfo info) { + public static void assertNoTaskSchedule(TaskSchedulingInfo info) { assertNotNull(info.getScheduledTaskNames()); assertTrue(info.getScheduledTaskNames().isEmpty()); + assertNoTaskErrors(info); + } + + public static void assertNoTaskErrors(TaskSchedulingInfo info) { assertNotNull(info.getGenerationErrors()); assertTrue(info.getGenerationErrors().isEmpty()); assertNotNull(info.getSchedulingErrors()); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java index b8833d10b1a1..dc396031f1e0 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java @@ -42,6 +42,7 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; @@ -408,18 +409,19 @@ public void testOfflineTableSingleLevelConcat() long expectedWatermark = 16000 * 86_400_000L; String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_CONCAT_TEST_TABLE); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(offlineTableName)); List taskList; - for (String tasks = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + for (String tasks = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + taskList = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNull(_taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -524,18 +526,19 @@ public void testOfflineTableSingleLevelConcatWithMetadataPush() long expectedWatermark = 16000 * 86_400_000L; String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(offlineTableName)); List taskList; - for (String tasks = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + for (String tasks = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + taskList = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNull(_taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -633,18 +636,19 @@ public void testOfflineTableSingleLevelRollup() long expectedWatermark = 16050 * 86_400_000L; String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_ROLLUP_TEST_TABLE); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(offlineTableName)); List taskList; - for (String tasks = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + for (String tasks = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + taskList = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), 1); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNull(_taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -785,18 +789,19 @@ public void testOfflineTableMultiLevelConcat() String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(MULTI_LEVEL_CONCAT_TEST_TABLE); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(offlineTableName)); List taskList; - for (String tasks = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + for (String tasks = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = _taskManager.scheduleAllTasksForTable(offlineTableName, null) + taskList = _taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNull(_taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -918,11 +923,12 @@ public void testRealtimeTableSingleLevelConcat() long expectedWatermark = 16000 * 86_400_000L; String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(tableName); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(realtimeTableName)); List taskList; - for (String tasks = taskManager.scheduleAllTasksForTable(realtimeTableName, null) + for (String tasks = taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = taskManager.scheduleAllTasksForTable(realtimeTableName, null) + taskList = taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { // assertEquals(helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); @@ -930,7 +936,7 @@ public void testRealtimeTableSingleLevelConcat() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(taskManager.scheduleAllTasksForTable(realtimeTableName, null) + assertNull(taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -1024,17 +1030,18 @@ public void testRealtimeTableProcessAllModeMultiLevelConcat() long[] expectedNumBucketsToProcess200Days = {0, 0, 1, 1, 0, 0, 1, 1}; String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(tableName); int numTasks = 0; + TaskSchedulingContext context = new TaskSchedulingContext(List.of(realtimeTableName)); List taskList; - for (String tasks = taskManager.scheduleAllTasksForTable(realtimeTableName, null). - get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = taskManager.scheduleAllTasksForTable(realtimeTableName, null) + for (String tasks = taskManager.scheduleTasks(context) + .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; + taskList = taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { assertTrue(helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - assertNull(taskManager.scheduleAllTasksForTable(realtimeTableName, null) + assertNull(taskManager.scheduleTasks(context) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); waitForTaskToComplete(); @@ -1066,10 +1073,10 @@ public void testRealtimeTableProcessAllModeMultiLevelConcat() uploadSegments(MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE, TableType.REALTIME, _tarDir5); waitForAllDocsLoaded(600_000L); - for (String tasks = taskManager.scheduleAllTasksForTable(realtimeTableName, null) + for (String tasks = taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames().get(0); tasks != null; - taskList = taskManager.scheduleAllTasksForTable(realtimeTableName, null) + taskList = taskManager.scheduleTasks(context) .get(MinionConstants.MergeRollupTask.TASK_TYPE).getScheduledTaskNames(), tasks = taskList != null && !taskList.isEmpty() ? taskList.get(0) : null, numTasks++) { waitForTaskToComplete(); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java index fed10b9f1ba5..919a87025905 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java @@ -33,6 +33,7 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.minion.MinionContext; import org.apache.pinot.spi.config.table.IndexingConfig; @@ -185,12 +186,13 @@ public void testFirstRunPurge() // 5. Check the purge process itself by setting an expecting number of rows String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(PURGE_FIRST_RUN_TABLE); - assertNotNull( - _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.PurgeTask.TASK_TYPE)); + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) + .get(MinionConstants.PurgeTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.PurgeTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains expected values @@ -200,7 +202,8 @@ public void testFirstRunPurge() metadata.getCustomMap().containsKey(MinionConstants.PurgeTask.TASK_TYPE + MinionConstants.TASK_TIME_SUFFIX)); } // Should not generate new purge task as the last time purge is not greater than last + 1day (default purge delay) - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); // 52 rows with ArrTime = 1 // 115545 totals rows @@ -231,11 +234,13 @@ public void testPassedDelayTimePurge() String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(PURGE_DELTA_PASSED_TABLE); assertNotNull( - _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.PurgeTask.TASK_TYPE)); + _taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) + .get(MinionConstants.PurgeTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.PurgeTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains expected values @@ -247,7 +252,8 @@ public void testPassedDelayTimePurge() assertTrue(System.currentTimeMillis() - Long.parseLong(purgeTime) < 86400000); } // Should not generate new purge task as the last time purge is not greater than last + 1day (default purge delay) - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); // 52 rows with ArrTime = 1 // 115545 totals rows @@ -279,7 +285,8 @@ public void testNotPassedDelayTimePurge() String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(PURGE_DELTA_NOT_PASSED_TABLE); // No task should be schedule as the delay is not passed - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { // Check purge time String purgeTime = @@ -330,11 +337,12 @@ public void testPurgeOnOldSegmentsWithIndicesOnNewColumns() // schedule purge tasks String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE); - assertNotNull( - _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.PurgeTask.TASK_TYPE)); + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) + .get(MinionConstants.PurgeTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.PurgeTask.TASK_TYPE))); - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.PurgeTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains expected values diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java index 296c981c1821..f15f16dba355 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java @@ -33,6 +33,7 @@ import org.apache.pinot.common.minion.RealtimeToOfflineSegmentsTaskMetadata; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.spi.config.table.ColumnPartitionConfig; import org.apache.pinot.spi.config.table.FieldConfig; @@ -231,13 +232,13 @@ public void testRealtimeToOfflineSegmentsTask() long expectedWatermark = _dataSmallestTimeMs + 86400000; for (int i = 0; i < 3; i++) { // Schedule task - assertNotNull(_taskManager.scheduleAllTasksForTable(_realtimeTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(_realtimeTableName))) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); assertTrue(_taskResourceManager.getTaskQueues().contains( PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE))); // Should not generate more tasks - MinionTaskTestUtils.assertNoTaskSchedule(_realtimeTableName, - MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule(new TaskSchedulingContext(_realtimeTableName, + MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE), _taskManager); // Wait at most 600 seconds for all tasks COMPLETED waitForTaskToComplete(expectedWatermark, _realtimeTableName); @@ -283,13 +284,13 @@ public void testRealtimeToOfflineSegmentsMetadataPushTask() _taskManager.cleanUpTask(); for (int i = 0; i < 3; i++) { // Schedule task - assertNotNull(_taskManager.scheduleAllTasksForTable(_realtimeMetadataTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(_realtimeTableName))) .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); assertTrue(_taskResourceManager.getTaskQueues().contains( PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE))); // Should not generate more tasks - MinionTaskTestUtils.assertNoTaskSchedule(_realtimeMetadataTableName, - MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule(new TaskSchedulingContext(_realtimeTableName, + MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE), _taskManager); // Wait at most 600 seconds for all tasks COMPLETED waitForTaskToComplete(expectedWatermark, _realtimeMetadataTableName); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RefreshSegmentMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RefreshSegmentMinionClusterIntegrationTest.java index c14f278cf6bd..daa455e841b2 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RefreshSegmentMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RefreshSegmentMinionClusterIntegrationTest.java @@ -33,6 +33,7 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.segment.spi.index.StandardIndexes; import org.apache.pinot.spi.config.table.FieldConfig; @@ -108,13 +109,13 @@ public void tearDown() throws Exception { public void testFirstSegmentRefresh() { // This will create the inverted index as we disable inverted index creation during segment push. String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(getTableName()); - assertNotNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) .get(MinionConstants.RefreshSegmentTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RefreshSegmentTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains expected values @@ -128,8 +129,8 @@ public void testFirstSegmentRefresh() { } // This should be no-op as nothing changes. - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { // Get the value in segment metadata Map customMap = metadata.getCustomMap(); @@ -153,13 +154,13 @@ public void testValidDatatypeChange() throws Exception { schema.getFieldSpecFor("DestAirportID").setDataType(FieldSpec.DataType.STRING); addSchema(schema); - assertNotNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) .get(MinionConstants.RefreshSegmentTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RefreshSegmentTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); waitForServerSegmentDownload(aVoid -> { @@ -232,13 +233,13 @@ public void testIndexChanges() throws Exception { updateTableConfig(tableConfig); String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(getTableName()); - assertNotNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) .get(MinionConstants.RefreshSegmentTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RefreshSegmentTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); waitForServerSegmentDownload(aVoid -> { @@ -323,13 +324,13 @@ public void checkColumnAddition() throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(getTableName()); - assertNotNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) .get(MinionConstants.RefreshSegmentTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RefreshSegmentTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains processed times. @@ -401,13 +402,13 @@ public void checkRefreshNotNecessary() throws Exception { updateTableConfig(tableConfig); - assertNotNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(offlineTableName))) .get(MinionConstants.RefreshSegmentTask.TASK_TYPE)); assertTrue(_helixTaskResourceManager.getTaskQueues() .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.RefreshSegmentTask.TASK_TYPE))); // Will not schedule task if there's incomplete task - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); waitForTaskToComplete(); // Check that metadata contains expected values @@ -423,8 +424,8 @@ public void checkRefreshNotNecessary() throws Exception { } // This should be no-op as nothing changes. - MinionTaskTestUtils.assertNoTaskSchedule(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE, - _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule( + new TaskSchedulingContext(offlineTableName, MinionConstants.RefreshSegmentTask.TASK_TYPE), _taskManager); for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { // Get the value in segment metadata Map customMap = metadata.getCustomMap(); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SimpleMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SimpleMinionClusterIntegrationTest.java index 3071d9c7fbc7..00dce3341c48 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SimpleMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SimpleMinionClusterIntegrationTest.java @@ -33,6 +33,7 @@ import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.minion.executor.PinotTaskExecutor; @@ -137,7 +138,7 @@ public void testStopResumeDeleteTaskQueue() { // Should create the task queues and generate a task in the same minion instance List task1 = - _taskManager.scheduleAllTasksForAllTables(null).get(TASK_TYPE).getScheduledTaskNames(); + _taskManager.scheduleTasks(new TaskSchedulingContext()).get(TASK_TYPE).getScheduledTaskNames(); assertNotNull(task1); assertEquals(task1.size(), 1); assertTrue(_helixTaskResourceManager.getTaskQueues() @@ -151,7 +152,7 @@ public void testStopResumeDeleteTaskQueue() { verifyTaskCount(task1.get(0), 0, 1, 1, 2); // Should generate one more task, with two sub-tasks. Both of these sub-tasks will wait // since we have one minion instance that is still running one of the sub-tasks. - List task2 = _taskManager.scheduleTaskForAllTables(TASK_TYPE, null).getScheduledTaskNames(); + List task2 = _taskManager.scheduleTasks(new TaskSchedulingContext()).get(TASK_TYPE).getScheduledTaskNames(); assertNotNull(task2); assertEquals(task2.size(), 1); assertTrue(_helixTaskResourceManager.getTasksInProgress(TASK_TYPE).contains(task2.get(0))); @@ -160,8 +161,7 @@ public void testStopResumeDeleteTaskQueue() { // Should not generate more tasks since SimpleMinionClusterIntegrationTests.NUM_TASKS is 2. // Our test task generator does not generate if there are already this many sub-tasks in the // running+waiting count already. - MinionTaskTestUtils.assertNoTaskSchedule(_taskManager); - MinionTaskTestUtils.assertNoTaskSchedule(TASK_TYPE, _taskManager); + MinionTaskTestUtils.assertNoTaskSchedule(new TaskSchedulingContext(), _taskManager); // Wait at most 60 seconds for all tasks IN_PROGRESS TestUtils.waitForCondition(input -> { diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java index e5c35f6dbd36..eceafc732c03 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TlsIntegrationTest.java @@ -58,6 +58,7 @@ import org.apache.pinot.common.utils.helix.HelixHelper; import org.apache.pinot.common.utils.tls.TlsUtils; import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.integration.tests.access.CertBasedTlsChannelAccessControlFactory; import org.apache.pinot.spi.config.table.TableConfig; @@ -478,7 +479,7 @@ public void testRealtimeSegmentUploadDownload() Assert.assertTrue(resultBeforeOffline.getResultSet(0).getLong(0) > 0); // schedule offline segment generation - Assert.assertNotNull(_controllerStarter.getTaskManager().scheduleAllTasksForAllTables(null)); + Assert.assertNotNull(_controllerStarter.getTaskManager().scheduleTasks(new TaskSchedulingContext())); // wait for offline segments JsonNode offlineSegments = TestUtils.waitForResult(() -> { diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java index 6d965ffae008..24d4b16d500e 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableIntegrationTest.java @@ -34,6 +34,7 @@ import org.apache.pinot.common.utils.config.TagNameUtils; import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager; import org.apache.pinot.core.data.manager.realtime.SegmentBuildTimeLeaseExtender; @@ -468,7 +469,7 @@ public void testUpsertCompaction() sendPostRequest(_controllerRequestURLBuilder.forResumeConsumption(tableName)); waitForNumQueriedSegmentsToConverge(tableName, 600_000L, 5, 2); String realtimeTableName = TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(tableName); - assertNotNull(_taskManager.scheduleAllTasksForTable(realtimeTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(realtimeTableName))) .get(MinionConstants.UpsertCompactionTask.TASK_TYPE)); waitForTaskToComplete(); // 2 segments should be compacted (351 rows -> 1 row; 500 rows -> 2 rows), 1 segment (149 rows) should be deleted @@ -501,7 +502,7 @@ public void testUpsertCompactionInMemory() // NOTE: When in-memory valid doc ids are used, no need to pause/resume consumption to trigger the snapshot. String realtimeTableName = TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(tableName); - assertNotNull(_taskManager.scheduleAllTasksForTable(realtimeTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(realtimeTableName))) .get(MinionConstants.UpsertCompactionTask.TASK_TYPE)); waitForTaskToComplete(); // 1 segment should be compacted (500 rows -> 2 rows) @@ -544,7 +545,7 @@ public void testUpsertCompactionWithSoftDelete() waitForNumQueriedSegmentsToConverge(tableName, 10_000L, 5, 2); String realtimeTableName = TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(tableName); - assertNotNull(_taskManager.scheduleAllTasksForTable(realtimeTableName, null) + assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext(List.of(realtimeTableName))) .get(MinionConstants.UpsertCompactionTask.TASK_TYPE)); waitForTaskToComplete(); // 1 segment should be compacted (351 rows -> 1 rows), 2 segments (500 rows, 151 rows) should be deleted diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java index 22d24115d975..ade04ba12e69 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UrlAuthRealtimeIntegrationTest.java @@ -32,6 +32,7 @@ import org.apache.pinot.client.ResultSetGroup; import org.apache.pinot.common.auth.UrlAuthProvider; import org.apache.pinot.controller.helix.ControllerRequestClient; +import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.spi.config.table.TableTaskConfig; import org.apache.pinot.spi.config.table.TableType; @@ -176,7 +177,7 @@ public void testSegmentUploadDownload() Assert.assertTrue(resultBeforeOffline.getResultSet(0).getLong(0) > 0); // schedule offline segment generation - Assert.assertNotNull(_controllerStarter.getTaskManager().scheduleAllTasksForAllTables(null)); + Assert.assertNotNull(_controllerStarter.getTaskManager().scheduleTasks(new TaskSchedulingContext())); // wait for offline segments List offlineSegments = TestUtils.waitForResult(() -> {