SbloodyS commented on code in PR #18358:
URL:
https://github.com/apache/dolphinscheduler/pull/18358#discussion_r3450148008
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java:
##########
@@ -199,46 +198,61 @@ public void buildK8sJob(K8sTaskMainParameters
k8STaskMainParameters) {
}
public void registerBatchJobWatcher(Job job, String taskInstanceId,
TaskResponse taskResponse) {
- CountDownLatch countDownLatch = new CountDownLatch(1);
- Watcher<Job> watcher = new Watcher<Job>() {
+ final String jobName = job.getMetadata().getName();
+ final String namespace = job.getMetadata().getNamespace();
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ SharedIndexInformer<Job> informer =
k8sUtils.createBatchJobInformer(jobName, namespace,
+ new ResourceEventHandler<Job>() {
+
+ @Override
+ public void onAdd(Job watchedJob) {
+ try {
+
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
+ taskRequest.getTaskInstanceId());
+
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
+ log.info("event received : job:{} action:ADD",
watchedJob.getMetadata().getName());
+ } finally {
+ LogUtils.removeTaskInstanceLogFullPathMDC();
+ LogUtils.removeWorkflowAndTaskInstanceIdMDC();
+ }
+ }
- @Override
- public void eventReceived(Action action, Job job) {
- try {
-
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
- taskRequest.getTaskInstanceId());
-
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
- log.info("event received : job:{} action:{}",
job.getMetadata().getName(), action);
- if (action == Action.DELETED) {
- log.error("[K8sJobExecutor-{}] fail in k8s",
job.getMetadata().getName());
- taskResponse.setExitStatusCode(EXIT_CODE_FAILURE);
- countDownLatch.countDown();
- } else if (action != Action.ADDED) {
- int jobStatus = getK8sJobStatus(job);
- log.info("job {} status {}",
job.getMetadata().getName(), jobStatus);
- if (jobStatus == TaskConstants.RUNNING_CODE) {
- return;
+ @Override
+ public void onUpdate(Job oldJob, Job watchedJob) {
+ try {
+
LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
+ taskRequest.getTaskInstanceId());
+
LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
+ log.info("event received : job:{} action:UPDATE",
watchedJob.getMetadata().getName());
+ int jobStatus = getK8sJobStatus(watchedJob);
+ log.info("job {} status {}",
watchedJob.getMetadata().getName(), jobStatus);
+ if (jobStatus == TaskConstants.RUNNING_CODE) {
+ return;
+ }
+ setTaskStatus(jobStatus, taskInstanceId,
taskResponse);
+ countDownLatch.countDown();
+ } finally {
+ LogUtils.removeTaskInstanceLogFullPathMDC();
+ LogUtils.removeWorkflowAndTaskInstanceIdMDC();
}
- setTaskStatus(jobStatus, taskInstanceId, taskResponse);
- countDownLatch.countDown();
}
- } finally {
- LogUtils.removeTaskInstanceLogFullPathMDC();
- LogUtils.removeWorkflowAndTaskInstanceIdMDC();
- }
- }
- @Override
Review Comment:
The old watcher handled `onClose(WatcherException)` by setting failure and
counting down. The new code waits indefinitely for non-timeout tasks. Consider
using `runnableInformer/start()` with a completion/error path, polling
`isRunning/isWatching`, or another explicit failure callback/monitor that
counts the latch down.
--
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]
