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

["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