merge from dev/4.2.x into main (#2420)

merge from dev/4.2.x into main
oceanbase · May 17, 2024 · 5bd8895 · 5bd8895
2 parents c7f5f28 + 9ef323b
commit 5bd8895
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 12 deletions.
diff --git a/client b/client
diff --git a/server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/TaskType.java b/server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/TaskType.java
@@ -114,4 +114,8 @@ public boolean needForExecutionStrategy() {
  return !(this == PRE_CHECK || this == SQL_CHECK || this == GENERATE_ROLLBACK);
  }
 
+ public boolean needsSetLogDownloadUrl() {
+ return !(this == PRE_CHECK || this == SQL_CHECK || this == GENERATE_ROLLBACK);
+ }
+
 }
diff --git a/...dc-service/src/main/java/com/oceanbase/odc/service/flow/task/BaseODCFlowTaskDelegate.java b/...dc-service/src/main/java/com/oceanbase/odc/service/flow/task/BaseODCFlowTaskDelegate.java
@@ -67,7 +67,6 @@
 import com.oceanbase.odc.service.task.model.OdcTaskLogLevel;
 
 import lombok.Getter;
-import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -126,7 +125,9 @@ private void initMonitorExecutor() {
  int interval = RuntimeTaskConstants.DEFAULT_TASK_CHECK_INTERVAL_SECONDS;
  scheduleExecutor.scheduleAtFixedRate(() -> {
  try {
- onProgressUpdate(taskId, taskService);
+ if (taskLatch.getCount() > 0) {
+ onProgressUpdate(taskId, taskService);
+ }
  } catch (Exception e) {
  log.warn("Update task progress callback failed, taskId={}", taskId, e);
  }
@@ -265,11 +266,14 @@ protected boolean isTimeout() {
  @Override
  public void callback(@NotNull long flowInstanceId, @NotNull long flowTaskInstanceId,
  @NotNull FlowNodeStatus flowNodeStatus, Map<String, Object> approvalVariables) {
- try {
- setDownloadLogUrl(flowInstanceId);
- } catch (Exception e) {
- log.warn("Failed to set download log URL, either because the log file does not exist "
- + "or the upload of the OSS failed, flowInstanceId={}", flowInstanceId, e);
+ if (getTaskType().needsSetLogDownloadUrl()) {
+ try {
+ setDownloadLogUrl();
+ } catch (Exception e) {
+ log.warn("Failed to set download log URL, either because the log file does not exist or the upload of "
+ + "the OSS failed, flowInstanceId={}, flowTaskInstanceId={}", flowInstanceId,
+ flowTaskInstanceId, e);
+ }
  }
  flowTaskCallBackApprovalService.approval(flowInstanceId, flowTaskInstanceId, flowNodeStatus, approvalVariables);
  }
@@ -336,7 +340,7 @@ public boolean cancel(boolean mayInterruptIfRunning) {
 
  protected abstract boolean cancel(boolean mayInterruptIfRunning, Long taskId, TaskService taskService);
 
- private void setDownloadLogUrl(@NonNull Long flowInstanceId) throws IOException, NotFoundException {
+ private void setDownloadLogUrl() throws IOException, NotFoundException {
  TaskEntity taskEntity = taskService.detail(taskId);
  File logFile;
  try {
@@ -346,9 +350,9 @@ private void setDownloadLogUrl(@NonNull Long flowInstanceId) throws IOException,
  // If the log file does not exist, the download URL will not be set
  return;
  }
- String downloadUrl = String.format("/api/v2/flow/flowInstances/%s/tasks/log/download", flowInstanceId);
+ String downloadUrl = String.format("/api/v2/flow/flowInstances/%s/tasks/log/download", getFlowInstanceId());
  if (Objects.nonNull(cloudObjectStorageService) && cloudObjectStorageService.supported()) {
- String fileName = TaskLogFilenameGenerator.generate(flowInstanceId);
+ String fileName = TaskLogFilenameGenerator.generate(getFlowInstanceId());
  try {
  String objectName = cloudObjectStorageService.uploadTemp(fileName, logFile);
  downloadUrl = TaskDownloadUrlsProvider

diff --git a/...ce/src/main/java/com/oceanbase/odc/service/flow/task/FlowTaskCallBackApprovalService.java b/...ce/src/main/java/com/oceanbase/odc/service/flow/task/FlowTaskCallBackApprovalService.java
@@ -103,7 +103,7 @@ private void completeTask(long flowInstanceId, long flowTaskInstanceId, FlowNode
  FlowInstanceEntity flowInstance = getFlowInstance(flowInstanceId);
  FlowableElement flowableElement = getFlowableElementOfUserTask(flowTaskInstanceId);
 
- Await.await().timeout(60).timeUnit(TimeUnit.SECONDS)
+ Await.await().timeout(60).timeUnit(TimeUnit.SECONDS).period(1).periodTimeUnit(TimeUnit.SECONDS)
  .until(getFlowableTask(flowInstance, flowableElement.getName())::isPresent).build().start();
  Task task = getFlowableTask(flowInstance, flowableElement.getName()).get();
  doCompleteTask(flowInstanceId, flowTaskInstanceId, flowNodeStatus, approvalVariables, task.getId());