kfaraz commented on code in PR #18587:
URL: https://github.com/apache/druid/pull/18587#discussion_r2404861989
##########
docs/development/extensions-core/k8s-jobs.md:
##########
@@ -792,7 +792,8 @@ Should you require the needed permissions for interacting
across Kubernetes name
| `druid.indexer.runner.graceTerminationPeriodSeconds` | `Long` | Number of
seconds you want to wait after a sigterm for container lifecycle hooks to
complete. Keep at a smaller value if you want tasks to hold locks for shorter
periods. | `PT30S` (K8s default) | No |
| `druid.indexer.runner.capacity` | `Integer` | Number of concurrent jobs that
can be sent to Kubernetes. | `2147483647` | No |
| `druid.indexer.runner.cpuCoreInMicro` | `Integer` | Number of CPU micro core
for the task. | `1000` | No |
-| `druid.indexer.runner.logSaveTimeout` | `Duration` | How long to wait for
task logs to be saved before giving up. | `PT300S` | NO |
+| `druid.indexer.runner.logSaveTimeout` | `Duration` | The peon executing the
ingestion task makes a best effort to persist the pod logs from `k8s` to
persistent task log storage. A timeout for this log persist is set to avoid
`k8s` connection issues causing the peon to hang indefinitely if issues occur
during the persisting of said logs. | `PT300S` | NO |
Review Comment:
```suggestion
| `druid.indexer.runner.logSaveTimeout` | `Duration` | The peon executing
the ingestion task makes a best effort to persist the pod logs from `k8s` to
persistent task log storage. The timeout ensures that `k8s` connection issues
do not cause the pod to hang indefinitely thereby blocking Overlord operations.
If the timeout occurs before the logs are saved, those logs will not be
available in Druid. | `PT300S` | NO |
```
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerConfig.java:
##########
@@ -157,14 +157,14 @@ private KubernetesTaskRunnerConfig(
Period taskCleanupDelay,
Period taskCleanupInterval,
Period k8sjobLaunchTimeout,
- Period logSaveTimeout,
List<String> peonMonitors,
List<String> javaOptsArray,
int cpuCoreInMicro,
Map<String, String> labels,
Map<String, String> annotations,
Integer capacity,
- Period taskJoinTimeout
+ Period taskJoinTimeout,
Review Comment:
Please revert the changes to this file since they are not needed anymore.
We can stick to the original order of the constructor args.
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -449,4 +449,35 @@ protected ListenableFuture<Boolean>
getTaskStartedSuccessfullyFuture()
{
return taskStartedSuccessfullyFuture;
}
+
+ /**
+ * Executes a callable with a timeout.
+ * <p>
+ * If the callable does not complete within the specified timeout or another
exception occurs, the error will be
+ * logged and null will be returned.
+ * </p>
+ */
+ private <T> @Nullable T executeWithTimeout(Callable<T> runnable, long
timeoutMillis, String operationName, String errorMessage)
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor(
+ Execs.makeThreadFactory("k8s-peon-lifecycle-util-" +
taskId.getOriginalTaskId() + "-%d"));
+ try {
+ Future<T> future = executor.submit(runnable);
+ return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.warn("Operation [%s] timed out after %d ms for task [%s]. %s",
operationName, timeoutMillis, taskId.getOriginalTaskId(), errorMessage);
Review Comment:
```suggestion
log.warn("Operation[%s] for task[%s] timed out after [%d] ms with
error[%s].", operationName, taskId.getOriginalTaskId(), timeoutMillis,
errorMessage);
```
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -449,4 +449,35 @@ protected ListenableFuture<Boolean>
getTaskStartedSuccessfullyFuture()
{
return taskStartedSuccessfullyFuture;
}
+
+ /**
+ * Executes a callable with a timeout.
+ * <p>
+ * If the callable does not complete within the specified timeout or another
exception occurs, the error will be
+ * logged and null will be returned.
+ * </p>
+ */
+ private <T> @Nullable T executeWithTimeout(Callable<T> runnable, long
timeoutMillis, String operationName, String errorMessage)
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor(
+ Execs.makeThreadFactory("k8s-peon-lifecycle-util-" +
taskId.getOriginalTaskId() + "-%d"));
+ try {
+ Future<T> future = executor.submit(runnable);
+ return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.warn("Operation [%s] timed out after %d ms for task [%s]. %s",
operationName, timeoutMillis, taskId.getOriginalTaskId(), errorMessage);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Operation [%s] was interrupted for task [%s]. %s",
operationName, taskId.getOriginalTaskId(), errorMessage);
Review Comment:
```suggestion
log.warn("Operation[%s] for task[%s] was interrupted with error[%s].",
operationName, taskId.getOriginalTaskId(), errorMessage);
```
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -449,4 +449,35 @@ protected ListenableFuture<Boolean>
getTaskStartedSuccessfullyFuture()
{
return taskStartedSuccessfullyFuture;
}
+
+ /**
+ * Executes a callable with a timeout.
+ * <p>
+ * If the callable does not complete within the specified timeout or another
exception occurs, the error will be
+ * logged and null will be returned.
+ * </p>
+ */
+ private <T> @Nullable T executeWithTimeout(Callable<T> runnable, long
timeoutMillis, String operationName, String errorMessage)
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor(
+ Execs.makeThreadFactory("k8s-peon-lifecycle-util-" +
taskId.getOriginalTaskId() + "-%d"));
+ try {
+ Future<T> future = executor.submit(runnable);
+ return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.warn("Operation [%s] timed out after %d ms for task [%s]. %s",
operationName, timeoutMillis, taskId.getOriginalTaskId(), errorMessage);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Operation [%s] was interrupted for task [%s]. %s",
operationName, taskId.getOriginalTaskId(), errorMessage);
+ }
+ catch (Exception e) {
+ log.error(e, "Error during operation [%s] for task: %s. %s",
operationName, taskId, errorMessage);
Review Comment:
```suggestion
log.error(e, "Error during operation[%s] for task[%s]: %s",
operationName, taskId, errorMessage);
```
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -449,4 +449,35 @@ protected ListenableFuture<Boolean>
getTaskStartedSuccessfullyFuture()
{
return taskStartedSuccessfullyFuture;
}
+
+ /**
+ * Executes a callable with a timeout.
+ * <p>
+ * If the callable does not complete within the specified timeout or another
exception occurs, the error will be
+ * logged and null will be returned.
+ * </p>
+ */
+ private <T> @Nullable T executeWithTimeout(Callable<T> runnable, long
timeoutMillis, String operationName, String errorMessage)
Review Comment:
```suggestion
private <T> @Nullable T executeWithTimeout(Callable<T> callable, long
timeoutMillis, String operationName, String errorMessage)
```
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -449,4 +449,35 @@ protected ListenableFuture<Boolean>
getTaskStartedSuccessfullyFuture()
{
return taskStartedSuccessfullyFuture;
}
+
+ /**
+ * Executes a callable with a timeout.
+ * <p>
+ * If the callable does not complete within the specified timeout or another
exception occurs, the error will be
+ * logged and null will be returned.
+ * </p>
+ */
+ private <T> @Nullable T executeWithTimeout(Callable<T> runnable, long
timeoutMillis, String operationName, String errorMessage)
+ {
+ ExecutorService executor = Executors.newSingleThreadExecutor(
+ Execs.makeThreadFactory("k8s-peon-lifecycle-util-" +
taskId.getOriginalTaskId() + "-%d"));
+ try {
+ Future<T> future = executor.submit(runnable);
+ return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.warn("Operation [%s] timed out after %d ms for task [%s]. %s",
operationName, timeoutMillis, taskId.getOriginalTaskId(), errorMessage);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
Review Comment:
@capistrant , regarding the previous discussion, I think we should avoid
marking this thread as interrupted here since we are not reading/clearing this
interrupted status anywhere which might have unintended side effects.
We can add it in the future if we need it.
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -97,7 +99,7 @@ protected enum State
private final ObjectMapper mapper;
private final TaskStateListener stateListener;
private final SettableFuture<Boolean> taskStartedSuccessfullyFuture;
- private final long logSaveTimeoutMs;
+ private final long logSaveTimeout;
Review Comment:
Nit: Please continue using `logSaveTimeoutMs` or `logSaveTimeoutMillis` as
it removes any ambiguity regarding the time unit.
Alternatively, you may pass in a `Duration` object and use the name
`logSaveTimeout`,
but I don't think that is needed 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: [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]
