capistrant commented on code in PR #18587:
URL: https://github.com/apache/druid/pull/18587#discussion_r2399034373
##########
extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesPeonLifecycle.java:
##########
@@ -343,52 +343,71 @@ protected void startWatchingLogs()
log.debug("There is already a log watcher for %s",
taskId.getOriginalTaskId());
return;
}
+ ExecutorService executor =
Executors.newSingleThreadExecutor(Execs.makeThreadFactory("k8s-tasklog-init-watch-%d"));
try {
- Optional<LogWatch> maybeLogWatch =
kubernetesClient.getPeonLogWatcher(taskId);
+ Future<Optional<LogWatch>> future = executor.submit(() ->
kubernetesClient.getPeonLogWatcher(taskId));
+ Optional<LogWatch> maybeLogWatch =
future.get(logWatchOperationTimeoutMs, TimeUnit.MILLISECONDS);
if (maybeLogWatch.isPresent()) {
logWatch = maybeLogWatch.get();
}
}
- catch (Exception e) {
- log.error(e, "Error watching logs from task: %s", taskId);
- }
- }
-
- protected void saveLogs()
- {
- ExecutorService executor =
Executors.newSingleThreadExecutor(Execs.makeThreadFactory("k8s-tasklog-persist-%d"));
- try {
- Future<?> future = executor.submit(this::doSaveLogs);
- future.get(logSaveTimeoutMs, TimeUnit.MILLISECONDS);
- }
catch (TimeoutException e) {
- log.warn("Persisting task logs timed out after %d ms for task [%s]. This
does not have any impact on the"
- + " work done by the task, but the logs may be innaccessible.
If this continues to happen, check"
- + " Kubernetes server logs for potential errors.",
logSaveTimeoutMs, taskId.getOriginalTaskId());
+ log.warn("Initializing log watcher timed out after %d ms for task [%s].
LogWatch not initialized.",
+ logWatchOperationTimeoutMs, taskId.getOriginalTaskId());
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Initializing log watcher was interrupted for task [%s].
LogWatch not initialized.", taskId.getOriginalTaskId());
}
catch (Exception e) {
- log.error(e, "Persisting task logs failed for task [%s] This does not
have any impact on the"
- + " work done by the task, but the logs may be
innaccessible. If this continues to happen, check"
- + " Kubernetes server logs for potential errors.",
taskId.getOriginalTaskId());
+ log.error(e, "Error watching logs from task: %s. LogWatch not
initialized.", taskId);
}
finally {
executor.shutdownNow();
- // shutdownNow does not always allow finally blocks to run, so we make
sure to close the logWatch here too if it
- // wasn't closed in doSaveLogs
- if (logWatch != null) {
- logWatch.close();
- }
}
}
- private void doSaveLogs()
+ protected void saveLogs()
Review Comment:
good point, I took a shot at this
--
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]
