Jackie-Jiang commented on a change in pull request #3819: Refactor periodic task URL: https://github.com/apache/incubator-pinot/pull/3819#discussion_r256193623
########## File path: pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java ########## @@ -44,134 +43,126 @@ * <p><code>PinotTaskManager</code> is also responsible for checking the health status on each type of tasks, detect and * fix issues accordingly. */ -public class PinotTaskManager extends ControllerPeriodicTask { +public class PinotTaskManager extends ControllerPeriodicTask<Void> { private static final Logger LOGGER = LoggerFactory.getLogger(PinotTaskManager.class); private final PinotHelixTaskResourceManager _helixTaskResourceManager; private final ClusterInfoProvider _clusterInfoProvider; private final TaskGeneratorRegistry _taskGeneratorRegistry; - private Map<String, List<TableConfig>> _enabledTableConfigMap; - private Set<String> _taskTypes; - private int _numTaskTypes; - private Map<String, String> _tasksScheduled; - - public PinotTaskManager(@Nonnull PinotHelixTaskResourceManager helixTaskResourceManager, - @Nonnull PinotHelixResourceManager helixResourceManager, @Nonnull ControllerConf controllerConf, - @Nonnull ControllerMetrics controllerMetrics) { + public PinotTaskManager(PinotHelixTaskResourceManager helixTaskResourceManager, + PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf, + ControllerMetrics controllerMetrics) { super("PinotTaskManager", controllerConf.getTaskManagerFrequencyInSeconds(), controllerConf.getPeriodicTaskInitialDelayInSeconds(), helixResourceManager, controllerMetrics); _helixTaskResourceManager = helixTaskResourceManager; _clusterInfoProvider = new ClusterInfoProvider(helixResourceManager, helixTaskResourceManager, controllerConf); _taskGeneratorRegistry = new TaskGeneratorRegistry(_clusterInfoProvider); } - @Override - protected void initTask() { - - } - /** - * Get the cluster info provider. - * <p>Cluster info provider might be needed to initialize task generators. + * Returns the cluster info provider. + * <p> + * Cluster info provider might be useful when initializing task generators. * * @return Cluster info provider */ - @Nonnull public ClusterInfoProvider getClusterInfoProvider() { return _clusterInfoProvider; } /** - * Register a task generator. - * <p>This is for pluggable task generators. + * Registers a task generator. + * <p> + * This method can be used to plug in custom task generators. * * @param pinotTaskGenerator Task generator to be registered */ - public void registerTaskGenerator(@Nonnull PinotTaskGenerator pinotTaskGenerator) { + public void registerTaskGenerator(PinotTaskGenerator pinotTaskGenerator) { _taskGeneratorRegistry.registerTaskGenerator(pinotTaskGenerator); } /** * Public API to schedule tasks. It doesn't matter whether current pinot controller is leader. */ - public Map<String, String> scheduleTasks() { - process(_pinotHelixResourceManager.getAllTables()); - return getTasksScheduled(); + public synchronized Map<String, String> scheduleTasks() { + Map<String, String> tasksScheduled = scheduleTasks(_pinotHelixResourceManager.getAllTables()); + + // For non-leader controller, perform non-leader cleanup + if (!isStarted()) { + nonLeaderCleanUp(); + } + return tasksScheduled; } - @Override - protected void preprocess() { - _metricsRegistry.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED, 1L); + /** + * Check the Pinot cluster status and schedule new tasks for the given tables. + * + * @param tableNamesWithType List of table names with type suffix + * @return Map from task type to task scheduled + */ + private Map<String, String> scheduleTasks(List<String> tableNamesWithType) { + _controllerMetrics.addMeteredGlobalValue(ControllerMeter.NUMBER_TIMES_SCHEDULE_TASKS_CALLED, 1L); - _taskTypes = _taskGeneratorRegistry.getAllTaskTypes(); - _numTaskTypes = _taskTypes.size(); - _enabledTableConfigMap = new HashMap<>(_numTaskTypes); + Set<String> taskTypes = _taskGeneratorRegistry.getAllTaskTypes(); + int numTaskTypes = taskTypes.size(); + Map<String, List<TableConfig>> enabledTableConfigMap = new HashMap<>(numTaskTypes); - for (String taskType : _taskTypes) { - _enabledTableConfigMap.put(taskType, new ArrayList<>()); + for (String taskType : taskTypes) { + enabledTableConfigMap.put(taskType, new ArrayList<>()); // Ensure all task queues exist _helixTaskResourceManager.ensureTaskQueueExists(taskType); } - } - @Override - protected void processTable(String tableNameWithType) { - TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType); - if (tableConfig != null) { - TableTaskConfig taskConfig = tableConfig.getTaskConfig(); - if (taskConfig != null) { - for (String taskType : _taskTypes) { - if (taskConfig.isTaskTypeEnabled(taskType)) { - _enabledTableConfigMap.get(taskType).add(tableConfig); + // Scan all table configs to get the tables with tasks enabled + for (String tableNameWithType : tableNamesWithType) { + TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType); + if (tableConfig != null) { + TableTaskConfig taskConfig = tableConfig.getTaskConfig(); + if (taskConfig != null) { + for (String taskType : taskTypes) { + if (taskConfig.isTaskTypeEnabled(taskType)) { + enabledTableConfigMap.get(taskType).add(tableConfig); + } } } } } - } - - @Override - protected void exceptionHandler(String tableNameWithType, Exception e) { - LOGGER.error("Exception in PinotTaskManager for table {}", tableNameWithType, e); - } - @Override - protected void postprocess() { // Generate each type of tasks - _tasksScheduled = new HashMap<>(_numTaskTypes); - for (String taskType : _taskTypes) { + Map<String, String> tasksScheduled = new HashMap<>(numTaskTypes); + for (String taskType : taskTypes) { LOGGER.info("Generating tasks for task type: {}", taskType); PinotTaskGenerator pinotTaskGenerator = _taskGeneratorRegistry.getTaskGenerator(taskType); - List<PinotTaskConfig> pinotTaskConfigs = pinotTaskGenerator.generateTasks(_enabledTableConfigMap.get(taskType)); + List<PinotTaskConfig> pinotTaskConfigs = pinotTaskGenerator.generateTasks(enabledTableConfigMap.get(taskType)); int numTasks = pinotTaskConfigs.size(); if (numTasks > 0) { LOGGER .info("Submitting {} tasks for task type: {} with task configs: {}", numTasks, taskType, pinotTaskConfigs); - _tasksScheduled.put(taskType, _helixTaskResourceManager + tasksScheduled.put(taskType, _helixTaskResourceManager .submitTask(pinotTaskConfigs, pinotTaskGenerator.getNumConcurrentTasksPerInstance())); - _metricsRegistry.addMeteredTableValue(taskType, ControllerMeter.NUMBER_TASKS_SUBMITTED, numTasks); + _controllerMetrics.addMeteredTableValue(taskType, ControllerMeter.NUMBER_TASKS_SUBMITTED, numTasks); } } - } - /** - * Returns the tasks that have been scheduled as part of the postprocess - * @return - */ - private Map<String, String> getTasksScheduled() { - return _tasksScheduled; + return tasksScheduled; } - /** - * Performs necessary cleanups (e.g. remove metrics) when the controller leadership changes. - */ - @Override - public void stopTask() { - LOGGER.info("Perform task cleanups."); - // Performs necessary cleanups for each task type. + private void nonLeaderCleanUp() { Review comment: Added comments in BasePeriodicTask ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: dev-unsubscr...@pinot.apache.org For additional commands, e-mail: dev-h...@pinot.apache.org