det101 commented on code in PR #18358:
URL:
https://github.com/apache/dolphinscheduler/pull/18358#discussion_r3450806768
##########
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:
Addressed: registered informer.stopped().whenComplete() to fail and count
down when informer stops with an exception (equivalent to old onClose). Only
triggers on ex != null to avoid false failure on normal stop(). Requires
Fabric8 6.4. Covered by testRegisterBatchJobInformerInformerStopped.
--
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]
