richardstartin commented on a change in pull request #8354:
URL: https://github.com/apache/pinot/pull/8354#discussion_r827873240
##########
File path:
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/TaskGeneratorUtils.java
##########
@@ -90,15 +90,48 @@ private TaskGeneratorUtils() {
if (isTaskOlderThanOneDay(taskName)) {
continue;
}
- for (PinotTaskConfig pinotTaskConfig :
clusterInfoAccessor.getTaskConfigs(entry.getKey())) {
+ for (PinotTaskConfig pinotTaskConfig :
clusterInfoAccessor.getTaskConfigs(taskName)) {
if
(tableNameWithType.equals(pinotTaskConfig.getConfigs().get(MinionConstants.TABLE_NAME_KEY)))
{
- nonCompletedTasks.put(entry.getKey(), entry.getValue());
+ nonCompletedTasks.put(taskName, entry.getValue());
}
}
}
return nonCompletedTasks;
}
+ /**
+ * Get all the tasks for the provided task type and tableName, which have
not reached final task state yet.
+ * In general, if the task is not in final state, we can treat it as
running, although it may wait to start
+ * or is paused. The caller provides a consumer to process the task configs
of those tasks.
+ */
+ public static void forRunningTasks(String tableNameWithType, String
taskType, ClusterInfoAccessor clusterInfoAccessor,
+ Consumer<Map<String, String>> taskConfigConsumer) {
+ Map<String, TaskState> taskStates =
clusterInfoAccessor.getTaskStates(taskType);
+ for (Map.Entry<String, TaskState> entry : taskStates.entrySet()) {
+ if (isTaskInFinalState(entry.getValue())) {
+ continue;
+ }
+ String taskName = entry.getKey();
+ for (PinotTaskConfig pinotTaskConfig :
clusterInfoAccessor.getTaskConfigs(taskName)) {
+ Map<String, String> config = pinotTaskConfig.getConfigs();
+ String tableNameFromTaskConfig =
config.get(MinionConstants.TABLE_NAME_KEY);
+ if (tableNameWithType.equals(tableNameFromTaskConfig)) {
+ taskConfigConsumer.accept(config);
+ }
+ }
+ }
+ }
+
+ /**
+ * @return true if task is in final state, i.e. will not be running any more.
+ * Note that STOPPED is not a final task state in helix task framework, as a
+ * stopped task is just paused and can be resumed to rerun.
+ */
+ private static boolean isTaskInFinalState(TaskState taskState) {
Review comment:
Can we replace this with an enumset and then replace the method call
above with finalstates.contains(state)
--
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]
