ADD log of the event

Author: luxiaolong-ct

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/k8s/impl/K8sTaskExecutor.java

@@ -206,12 +206,35 @@ public void registerBatchJobWatcher(Job job, String taskInstanceId, TaskResponse
 
                     @Override
                     public void onAdd(Job watchedJob) {
-                        // ignore initial add event, same as Watcher.Action.ADDED
+                        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 onUpdate(Job oldJob, Job watchedJob) {
-                        handleBatchJobEvent(watchedJob, taskInstanceId, taskResponse, countDownLatch);
+                        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();
+                        }
                     }
 
                     @Override
@@ -250,28 +273,6 @@ public void onDelete(Job watchedJob, boolean deletedFinalStateUnknown) {
         }
     }
 
-    private void handleBatchJobEvent(Job watchedJob,
-                                     String taskInstanceId,
-                                     TaskResponse taskResponse,
-                                     CountDownLatch countDownLatch) {
-        try {
-            LogUtils.setWorkflowAndTaskInstanceIDMDC(taskRequest.getWorkflowInstanceId(),
-                    taskRequest.getTaskInstanceId());
-            LogUtils.setTaskInstanceLogFullPathMDC(taskRequest.getLogPath());
-            log.info("event received : job:{}", 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();
-        }
-    }
-
     private void parsePodLogOutput() {
         ExecutorService collectPodLogExecutorService = ThreadUtils
                 .newSingleDaemonScheduledExecutorService("CollectPodLogOutput-thread-" + taskRequest.getTaskName());