You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "nozjkoitop (via GitHub)" <gi...@apache.org> on 2025/05/27 14:08:04 UTC

[PR] Configurable timeout for index_parallel subtasks (druid)

nozjkoitop opened a new pull request, #18039:
URL: https://github.com/apache/druid/pull/18039

   ### Description
   
   Introduce a runtime limit on each index_parallel subtask. Once a subtask starts, a timer kicks off. If the subtask runs longer than the configured window, it’s force-cancelled and the parent task fails.
   
   `subTaskTimeoutMs` property is added to the `tuning config`
   
   This PR has:
   
   - [X] been self-reviewed.
   - [X] added documentation for new or modified features or behaviors.
   - [ ] a release note entry in the PR description.
   - [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
   - [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
   - [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
   - [X] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
   - [ ] added integration tests.
   - [X] been tested in a test Druid cluster.
   


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18039:
URL: https://github.com/apache/druid/pull/18039#issuecomment-2914760647

   @nozjkoitop , could you please share some details on how this feature can be helpful in daily Druid operation?
   Given that this just causes the parent task to fail, I am not sure what value it adds other than probably freeing up task slots.
   Is that the sole intention here?


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18039:
URL: https://github.com/apache/druid/pull/18039#issuecomment-2915653745

   Thanks for the details, @nozjkoitop , could you please add the same to the PR description itself?
   
   Suggestions for this PR:
   - Rather than adding a new config, we should add this as a new task context parameter.
   - Use the prefix `Millis` rather than `Ms`.
   - Add documentation for the new task context parameter


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121081850


##########
docs/ingestion/tasks.md:
##########
@@ -472,6 +472,7 @@ The following parameters apply to all task types.
 |`useLineageBasedSegmentAllocation`|Enables the new lineage-based segment allocation protocol for the native Parallel task with dynamic partitioning. This option should be off during the replacing rolling upgrade from one of the Druid versions between 0.19 and 0.21 to Druid 0.22 or higher. Once the upgrade is done, it must be set to `true` to ensure data correctness.|`false` in 0.21 or earlier, `true` in 0.22 or later|
 |`lookupLoadingMode`|Controls the lookup loading behavior in tasks. This property supports three values: `ALL` mode loads all the lookups, `NONE` mode does not load any lookups and `ONLY_REQUIRED` mode loads the lookups specified with context key `lookupsToLoad`. This property must not be specified for `MSQ` and `kill` tasks as the task engine enforces `ONLY_REQUIRED` mode for `MSQWorkerTask` and `NONE` mode for `MSQControllerTask` and `kill` tasks.|`ALL`|
 |`lookupsToLoad`|List of lookup names to load in tasks. This property is required only if the `lookupLoadingMode` is set to `ONLY_REQUIRED`. For `MSQWorkerTask` type, the lookup names to load are identified by the controller task by parsing the SQL. |`null`|
+|`subTaskTimeoutMillis`|Maximum time (in milliseconds) to wait before cancelling a long-running worker task. Set to 0 for no timeout (infinite).|0 (unlimited)|

Review Comment:
   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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121192558


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);
+                  FutureUtils.getUnchecked(overlordClient.cancelTask(taskId), true);
+                  final TaskStatusPlus cancelledTaskStatus = FutureUtils.getUnchecked(
+                          overlordClient.taskStatus(taskId), true).getStatus();
+                  reportsMap.remove(taskId);
+                  incrementNumFailedTasks();
+
+                  if (monitorEntry.numTries() < maxRetry) {
+                    retry(specId, monitorEntry, cancelledTaskStatus);

Review Comment:
   Not in case of timeouts caused by occasional deadlocks, than it might be beneficial



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #18039:
URL: https://github.com/apache/druid/pull/18039


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121198505


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;

Review Comment:
   Fair, thanks, removed now



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121085086


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java:
##########
@@ -472,4 +474,18 @@ int getAndIncrementNextSpecId()
   {
     return nextSpecId++;
   }
+
+  private long getSubtaskTimeoutMillisFromContext()
+  {
+    if (context.isEmpty()) {
+      return Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;
+    }
+
+    Object raw = context.getOrDefault(
+            Tasks.SUB_TASK_TIMEOUT_KEY,
+            Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS
+    );
+
+    return (raw instanceof Number) ? ((Number) raw).longValue() : Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;

Review Comment:
   In terms of <String, Object> map we can receive a CCE on the Integer value, so i'd go with the cast to number here 



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java:
##########
@@ -48,6 +48,7 @@ public class ParallelIndexTuningConfig extends IndexTuningConfig
   private static final int DEFAULT_MAX_NUM_SEGMENTS_TO_MERGE = 100;
   private static final int DEFAULT_TOTAL_NUM_MERGE_TASKS = 10;
   private static final int DEFAULT_MAX_ALLOWED_LOCK_COUNT = -1;
+  private static final int DEFAULT_SUBTASK_TIMEOUT = 0;

Review Comment:
   Thanks for noticing



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121149653


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java:
##########
@@ -472,4 +474,18 @@ int getAndIncrementNextSpecId()
   {
     return nextSpecId++;
   }
+
+  private long getSubtaskTimeoutMillisFromContext()
+  {
+    if (context.isEmpty()) {
+      return Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;
+    }
+
+    Object raw = context.getOrDefault(
+            Tasks.SUB_TASK_TIMEOUT_KEY,
+            Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS
+    );
+
+    return (raw instanceof Number) ? ((Number) raw).longValue() : Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;

Review Comment:
   Sure, that would work too.



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121201587


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);
+                  FutureUtils.getUnchecked(overlordClient.cancelTask(taskId), true);
+                  final TaskStatusPlus cancelledTaskStatus = FutureUtils.getUnchecked(
+                          overlordClient.taskStatus(taskId), true).getStatus();
+                  reportsMap.remove(taskId);
+                  incrementNumFailedTasks();
+
+                  if (monitorEntry.numTries() < maxRetry) {
+                    retry(specId, monitorEntry, cancelledTaskStatus);

Review Comment:
   Do you mean a deadlock in the code?
   
   In that case, I think that failing the tasks is even more important.
   Otherwise, we may go on retrying silently and never catch the problem in the code.
   
   I would advise that we avoid doing the retries in the first iteration of this config,
   and treat the `subTaskTimeoutMillis` as a hard stop for any given sub-task.
   
   Let me know what you think.



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121085821


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +135,26 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long now = System.currentTimeMillis();

Review Comment:
   Updated 



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on PR #18039:
URL: https://github.com/apache/druid/pull/18039#issuecomment-2937317768

   Is this one ok to merge? The failures look to me like the ones expected to be fixed in #18067.


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on PR #18039:
URL: https://github.com/apache/druid/pull/18039#issuecomment-2915533309

   Hi @kfaraz 
   We’ve had multiple incidents where subtasks hung for days doing nothing and just occupying slots. Bugs got fixed, but new ones can always pop up. Adding subTaskTimeoutMs gives us a simple safeguard.
   
   ![image](https://github.com/user-attachments/assets/aaeac301-0387-4f96-a193-a7adfaa25e57)
   
   Key benefits:
   1. Fail fast: instead of waiting forever, a stuck subtask is cancelled and the parent job errors immediately.
   2. Free slots: zombies don’t tie up worker slots for hours or days.
   3. Automate recovery: failures trigger existing retry/alert pipelines, so ops can fix or let automation spin up a fresh attempt.


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121260233


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);
+                  FutureUtils.getUnchecked(overlordClient.cancelTask(taskId), true);
+                  final TaskStatusPlus cancelledTaskStatus = FutureUtils.getUnchecked(
+                          overlordClient.taskStatus(taskId), true).getStatus();
+                  reportsMap.remove(taskId);
+                  incrementNumFailedTasks();
+
+                  if (monitorEntry.numTries() < maxRetry) {
+                    retry(specId, monitorEntry, cancelledTaskStatus);

Review Comment:
   I agree, one might want to bring it back partially in future, for those edge cases we know are truly transient.



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "nozjkoitop (via GitHub)" <gi...@apache.org>.
nozjkoitop commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121260233


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);
+                  FutureUtils.getUnchecked(overlordClient.cancelTask(taskId), true);
+                  final TaskStatusPlus cancelledTaskStatus = FutureUtils.getUnchecked(
+                          overlordClient.taskStatus(taskId), true).getStatus();
+                  reportsMap.remove(taskId);
+                  incrementNumFailedTasks();
+
+                  if (monitorEntry.numTries() < maxRetry) {
+                    retry(specId, monitorEntry, cancelledTaskStatus);

Review Comment:
   Agree, might want to bring it back partially in future, for those edge cases we know are truly transient.



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18039:
URL: https://github.com/apache/druid/pull/18039#issuecomment-2938663734

   Thanks for the contribution, @nozjkoitop , @sviatahorau.


-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2120922857


##########
docs/ingestion/tasks.md:
##########
@@ -472,6 +472,7 @@ The following parameters apply to all task types.
 |`useLineageBasedSegmentAllocation`|Enables the new lineage-based segment allocation protocol for the native Parallel task with dynamic partitioning. This option should be off during the replacing rolling upgrade from one of the Druid versions between 0.19 and 0.21 to Druid 0.22 or higher. Once the upgrade is done, it must be set to `true` to ensure data correctness.|`false` in 0.21 or earlier, `true` in 0.22 or later|
 |`lookupLoadingMode`|Controls the lookup loading behavior in tasks. This property supports three values: `ALL` mode loads all the lookups, `NONE` mode does not load any lookups and `ONLY_REQUIRED` mode loads the lookups specified with context key `lookupsToLoad`. This property must not be specified for `MSQ` and `kill` tasks as the task engine enforces `ONLY_REQUIRED` mode for `MSQWorkerTask` and `NONE` mode for `MSQControllerTask` and `kill` tasks.|`ALL`|
 |`lookupsToLoad`|List of lookup names to load in tasks. This property is required only if the `lookupLoadingMode` is set to `ONLY_REQUIRED`. For `MSQWorkerTask` type, the lookup names to load are identified by the controller task by parsing the SQL. |`null`|
+|`subTaskTimeoutMillis`|Maximum time (in milliseconds) to wait before cancelling a long-running worker task. Set to 0 for no timeout (infinite).|0 (unlimited)|

Review Comment:
   ```suggestion
   |`subTaskTimeoutMillis`|Maximum time (in milliseconds) to wait before cancelling a long-running sub-task. Applicable only for `index_parallel` tasks and `compact` tasks (when running in parallel mode). Set to 0 for no timeout (infinite).|0 (unlimited)|
   ```



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java:
##########
@@ -472,4 +474,18 @@ int getAndIncrementNextSpecId()
   {
     return nextSpecId++;
   }
+
+  private long getSubtaskTimeoutMillisFromContext()
+  {
+    if (context.isEmpty()) {
+      return Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;
+    }
+
+    Object raw = context.getOrDefault(
+            Tasks.SUB_TASK_TIMEOUT_KEY,
+            Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS
+    );
+
+    return (raw instanceof Number) ? ((Number) raw).longValue() : Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;

Review Comment:
   We should throw an exception if the given value is not a Long.
   So, we can probably just simplify this:
   
   ```suggestion
       return (long) context.getOrDefault(
               Tasks.SUB_TASK_TIMEOUT_KEY,
               Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS
       );
   ```
   



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java:
##########
@@ -472,4 +474,18 @@ int getAndIncrementNextSpecId()
   {
     return nextSpecId++;
   }
+
+  private long getSubtaskTimeoutMillisFromContext()
+  {
+    if (context.isEmpty()) {
+      return Tasks.DEFAULT_SUB_TASK_TIMEOUT_MILLIS;
+    }

Review Comment:
   This is unnecessary since we are already using a default in the subsequent code.



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java:
##########
@@ -48,6 +48,7 @@ public class ParallelIndexTuningConfig extends IndexTuningConfig
   private static final int DEFAULT_MAX_NUM_SEGMENTS_TO_MERGE = 100;
   private static final int DEFAULT_TOTAL_NUM_MERGE_TASKS = 10;
   private static final int DEFAULT_MAX_ALLOWED_LOCK_COUNT = -1;
+  private static final int DEFAULT_SUBTASK_TIMEOUT = 0;

Review Comment:
   Not needed anymore.



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +135,26 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long now = System.currentTimeMillis();

Review Comment:
   Maybe use a `Stopwatch` instead of `System.currentTimeMillis`.



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -88,6 +88,7 @@ public class TaskMonitor<T extends Task, SubTaskReportType extends SubTaskReport
   private final OverlordClient overlordClient;
   private final int maxRetry;
   private final int estimatedNumSucceededTasks;
+  private final long taskTimeoutMs;

Review Comment:
   ```suggestion
     private final long taskTimeoutMillis;
   ```



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

Re: [PR] Configurable timeout for index_parallel subtasks (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18039:
URL: https://github.com/apache/druid/pull/18039#discussion_r2121156148


##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -106,14 +108,14 @@ public class TaskMonitor<T extends Task, SubTaskReportType extends SubTaskReport
   @GuardedBy("startStopLock")
   private boolean running = false;
 
-  TaskMonitor(OverlordClient overlordClient, int maxRetry, int estimatedNumSucceededTasks)
+  TaskMonitor(OverlordClient overlordClient, int maxRetry, int estimatedNumSucceededTasks, long taskTimeoutMillis)
   {
     // Unlimited retries for Overlord APIs: if it goes away, we'll wait indefinitely for it to come back.
     this.overlordClient = Preconditions.checkNotNull(overlordClient, "overlordClient")
                                        .withRetryPolicy(StandardRetryPolicy.unlimited());
     this.maxRetry = maxRetry;
     this.estimatedNumSucceededTasks = estimatedNumSucceededTasks;
-
+    this.taskTimeoutMillis = taskTimeoutMillis;
     log.info("TaskMonitor is initialized with estimatedNumSucceededTasks[%d]", estimatedNumSucceededTasks);

Review Comment:
   ```suggestion
       log.info("TaskMonitor is initialized with estimatedNumSucceededTasks[%d] and sub-task timeout[%d millis].", estimatedNumSucceededTasks, taskTimeoutMillis);
   ```



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java:
##########
@@ -275,6 +275,7 @@ public int getMaxAllowedLockCount()
     return maxAllowedLockCount;
   }
 
+

Review Comment:
   Extra new line:
   ```suggestion
   ```



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;

Review Comment:
   This assignment seems unncessary, we can just use `taskTimeoutMillis` itself.



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);
+                  FutureUtils.getUnchecked(overlordClient.cancelTask(taskId), true);
+                  final TaskStatusPlus cancelledTaskStatus = FutureUtils.getUnchecked(
+                          overlordClient.taskStatus(taskId), true).getStatus();
+                  reportsMap.remove(taskId);
+                  incrementNumFailedTasks();
+
+                  if (monitorEntry.numTries() < maxRetry) {
+                    retry(specId, monitorEntry, cancelledTaskStatus);

Review Comment:
   I am not sure if we should retry a task that has been cancelled for exceeding the timeout.
   Doesn't retrying somewhat defeat the purpose of the timeout?



##########
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TaskMonitor.java:
##########
@@ -134,6 +136,25 @@ public void start(long taskStatusCheckingPeriod)
                 final MonitorEntry monitorEntry = entry.getValue();
                 final String taskId = monitorEntry.runningTask.getId();
 
+                final long timeout = taskTimeoutMillis;
+                final long elapsed = monitorEntry.getStopwatch().millisElapsed();
+                if (timeout > 0 && elapsed > timeout) {
+                  log.warn("task[%s] timed out after %s ms, cancelling...", taskId, elapsed);

Review Comment:
   ```suggestion
                     log.warn("Cancelling task[%s] as it has already run for [%d] millis (taskTimeoutMillis=[%d]).", taskId, elapsed, taskTimeoutMillis);
   ```



-- 
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@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org