fix(taskframework): fix retry log attribute (#1904)

Author: krihy

server/odc-service/src/main/java/com/oceanbase/odc/service/task/constants/JobAttributeEntityColumn.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.oceanbase.odc.service.task.constants;
+
+/**
+ * @author yaobin
+ * @date 2024-03-13
+ * @since 4.2.4
+ */
+public class JobAttributeEntityColumn {
+
+    public static final String ID = "id";
+
+    public static final String ATTRIBUTE_KEY = "attributeKey";
+
+    public static final String ATTRIBUTE_VALUE = "attributeValue";
+
+}

server/odc-service/src/main/java/com/oceanbase/odc/service/task/executor/task/BaseTask.java

@@ -59,7 +59,7 @@ public void start(JobContext context) {
                 updateStatus(JobStatus.FAILED);
             }
         } catch (Throwable e) {
-            log.info("Task failed, id={}.", getJobId(), e);
+            log.warn("Task failed, id={}.", getJobId(), e);
             updateStatus(JobStatus.FAILED);
         } finally {
             try {

server/odc-service/src/main/java/com/oceanbase/odc/service/task/service/StdTaskFrameworkService.java

@@ -57,6 +57,7 @@
 import com.oceanbase.odc.metadb.task.JobEntity;
 import com.oceanbase.odc.metadb.task.JobRepository;
 import com.oceanbase.odc.service.task.config.TaskFrameworkProperties;
+import com.oceanbase.odc.service.task.constants.JobAttributeEntityColumn;
 import com.oceanbase.odc.service.task.constants.JobEntityColumn;
 import com.oceanbase.odc.service.task.enums.JobStatus;
 import com.oceanbase.odc.service.task.enums.TaskRunMode;
@@ -315,14 +316,34 @@ private void updateJobScheduleEntity(TaskResult taskResult) {
         jobRepository.update(jse);
 
         if (taskResult.getLogMetadata() != null && taskResult.getStatus().isTerminated()) {
-            taskResult.getLogMetadata().forEach((k, v) -> {
+            saveOrUpdateLogMetadata(taskResult, jse);
+        }
+    }
+
+    private void saveOrUpdateLogMetadata(TaskResult taskResult, JobEntity jse) {
+        taskResult.getLogMetadata().forEach((k, v) -> {
+            // log key may exist if job is retrying
+            Optional<String> logValue = findByJobIdAndAttributeKey(jse.getId(), k);
+            if (logValue.isPresent()) {
+                updateJobAttributeValue(jse.getId(), k, v);
+            } else {
                 JobAttributeEntity jobAttribute = new JobAttributeEntity();
                 jobAttribute.setJobId(jse.getId());
                 jobAttribute.setAttributeKey(k);
                 jobAttribute.setAttributeValue(v);
                 jobAttributeRepository.save(jobAttribute);
-            });
-        }
+            }
+        });
+    }
+
+    private void updateJobAttributeValue(Long id, String key, String value) {
+        CriteriaBuilder cb = entityManager.getCriteriaBuilder();
+        CriteriaUpdate<JobAttributeEntity> update = cb.createCriteriaUpdate(JobAttributeEntity.class);
+        Root<JobAttributeEntity> e = update.from(JobAttributeEntity.class);
+        update.set(JobAttributeEntityColumn.ATTRIBUTE_VALUE, value);
+        update.where(cb.equal(e.get(JobAttributeEntityColumn.ID), id),
+                cb.equal(e.get(JobAttributeEntityColumn.ATTRIBUTE_KEY), key));
+        entityManager.createQuery(update).executeUpdate();
     }
 
     @Transactional(rollbackFor = Exception.class)
@@ -467,4 +488,5 @@ public Optional<String> findByJobIdAndAttributeKey(Long jobId, String attributeK
         JobAttributeEntity attributeEntity = jobAttributeRepository.findByJobIdAndAttributeKey(jobId, attributeKey);
         return Objects.isNull(attributeEntity) ? Optional.empty() : Optional.of(attributeEntity.getAttributeValue());
     }
+
 }