SbloodyS commented on code in PR #18358:
URL:
https://github.com/apache/dolphinscheduler/pull/18358#discussion_r3450137983
##########
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());
Review Comment:
```suggestion
log.info("event received, job: {}, action: ADD",
watchedJob.getMetadata().getName());
```
##########
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) {
Review Comment:
You should also evaluate terminal Job status and count down.
##########
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());
Review Comment:
```suggestion
log.info("event received, job: {}, action:
UPDATE", watchedJob.getMetadata().getName());
```
##########
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. 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]
