merge 202502bp into main

merge 202502bp into main

Author: LioRoger

pom.xml

@@ -782,11 +782,6 @@
                 <artifactId>commons-compress</artifactId>
                 <version>${commons-compress.version}</version>
             </dependency>
-            <dependency>
-                <groupId>commons-codec</groupId>
-                <artifactId>commons-codec</artifactId>
-                <version>1.14</version>
-            </dependency>
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-collections4</artifactId>

server/odc-common/src/main/java/com/oceanbase/odc/common/task/RouteLogCallable.java

@@ -22,7 +22,7 @@
 import org.slf4j.MDC;
 
 public abstract class RouteLogCallable<T> implements Callable<T> {
-
+    public static final String LOG_PATH_PATTERN = "%s/%s/%s/%s.log";
     protected static Logger log = LogManager.getLogger(RouteLogCallable.class);
     private final String workSpace;
     private final String taskId;

server/odc-core/src/main/java/com/oceanbase/odc/core/alarm/AlarmUtils.java

@@ -27,6 +27,7 @@ public final class AlarmUtils {
      * Base alarm message names
      */
     public static final String CLUSTER_NAME = "Cluster";
+    public static final String INSTANCE_NAME = "instanceId";
     public static final String TENANT_NAME = "Tenant";
     public static final String ORGANIZATION_NAME = "OrganizationId";
     public static final String MESSAGE_NAME = "Message";
@@ -42,6 +43,7 @@ public final class AlarmUtils {
     public static final String RESOURCE_TYPE = "ResourceType";
     public static final String TASK_TYPE_NAME = "TaskType";
     public static final String SCHEDULE_ID_NAME = "ScheduleId";
+    public static final String FLOW_INSTANCE_ID_NAME = "FlowInstanceId";
     public static final Collection<String> TASK_FRAMEWORK_ALARM_DIGEST_NAMES =
             Arrays.asList(CLUSTER_NAME, TENANT_NAME, SCHEDULE_ID_NAME);
 

server/odc-server/src/main/java/com/oceanbase/odc/server/web/controller/v2/ExportController.java

@@ -15,6 +15,8 @@
  */
 package com.oceanbase.odc.server.web.controller.v2;
 
+import java.util.List;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -25,7 +27,10 @@
 import com.oceanbase.odc.service.common.response.SuccessResponse;
 import com.oceanbase.odc.service.schedule.export.ScheduleExportService;
 import com.oceanbase.odc.service.schedule.export.model.FileExportResponse;
+import com.oceanbase.odc.service.schedule.export.model.ScheduleExportListView;
 import com.oceanbase.odc.service.schedule.export.model.ScheduleTaskExportRequest;
+import com.oceanbase.odc.service.state.model.StateName;
+import com.oceanbase.odc.service.state.model.StatefulRoute;
 
 @RequestMapping("/api/v2/export")
 @RestController
@@ -34,8 +39,27 @@ public class ExportController {
     @Autowired
     private ScheduleExportService scheduleExportService;
 
-    @RequestMapping(value = "/exportScheduleTask", method = RequestMethod.POST)
-    public SuccessResponse<FileExportResponse> exportScheduleTask(@RequestBody ScheduleTaskExportRequest request) {
-        return Responses.success(scheduleExportService.export(request));
+    @RequestMapping(value = "getExportListView", method = RequestMethod.POST)
+    public SuccessResponse<List<ScheduleExportListView>> getExportListView(
+            @RequestBody ScheduleTaskExportRequest request) {
+        return Responses.success(scheduleExportService.getExportListView(request));
+    }
+
+    @RequestMapping(value = "/exportSchedule", method = RequestMethod.POST)
+    public SuccessResponse<String> exportScheduleTask2(@RequestBody ScheduleTaskExportRequest request) {
+        return Responses.success(scheduleExportService.startExport(request));
+    }
+
+    @RequestMapping(value = "/getExportResult", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#exportId")
+    public SuccessResponse<FileExportResponse> exportScheduleTask(String exportId) {
+        return Responses.success(scheduleExportService.getExportResult(exportId));
+    }
+
+    @RequestMapping(value = "/getExportLog", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#exportId")
+    public SuccessResponse<String> getExportLog(String exportId) {
+        return Responses.success(scheduleExportService.getExportLog(exportId));
     }
+
 }

server/odc-server/src/main/java/com/oceanbase/odc/server/web/controller/v2/FlowInstanceController.java

@@ -16,6 +16,7 @@
 package com.oceanbase.odc.server.web.controller.v2;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
@@ -49,6 +50,7 @@
 import com.oceanbase.odc.service.common.util.WebResponseUtils;
 import com.oceanbase.odc.service.flow.FlowInstanceService;
 import com.oceanbase.odc.service.flow.FlowTaskInstanceService;
+import com.oceanbase.odc.service.flow.model.BatchTerminateFlowResult;
 import com.oceanbase.odc.service.flow.model.BinaryDataResult;
 import com.oceanbase.odc.service.flow.model.CreateFlowInstanceReq;
 import com.oceanbase.odc.service.flow.model.FlowInstanceApprovalReq;
@@ -61,6 +63,8 @@
 import com.oceanbase.odc.service.partitionplan.model.PartitionPlanConfig;
 import com.oceanbase.odc.service.schedule.ScheduleService;
 import com.oceanbase.odc.service.session.model.SqlExecuteResult;
+import com.oceanbase.odc.service.state.model.StateName;
+import com.oceanbase.odc.service.state.model.StatefulRoute;
 import com.oceanbase.odc.service.task.model.OdcTaskLogLevel;
 
 import io.swagger.annotations.ApiOperation;
@@ -251,4 +255,24 @@ public SuccessResponse<PartitionPlanConfig> getPartitionPlan(@PathVariable Long
         return Responses.ok(this.partitionPlanScheduleService.getPartitionPlanByFlowInstanceId(id));
     }
 
+    @ApiOperation(value = "cancelFlowInstance", notes = "批量终止流程")
+    @RequestMapping(value = "/asyncCancel", method = RequestMethod.POST)
+    public SuccessResponse<String> batchCancelFlowInstance(@RequestBody Collection<Long> flowInstanceIds) {
+        return Responses.single(flowInstanceService.startBatchCancelFlowInstance(flowInstanceIds));
+    }
+
+    @ApiOperation(value = "getBatchCancelResult", notes = "获取批量终止结果")
+    @RequestMapping(value = "/asyncCancelResult", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#terminateId")
+    public SuccessResponse<List<BatchTerminateFlowResult>> getBatchCancelResult(String terminateId) {
+        return Responses.single(flowInstanceService.getBatchCancelResult(terminateId));
+    }
+
+    @ApiOperation(value = "getBatchCancelLog", notes = "获取批量终止日志")
+    @RequestMapping(value = "/asyncCancelLog", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#terminateId")
+    public SuccessResponse<String> getBatchCancelLog(String terminateId) {
+        return Responses.single(flowInstanceService.getBatchCancelLog(terminateId));
+    }
+
 }

server/odc-server/src/main/java/com/oceanbase/odc/server/web/controller/v2/ScheduleController.java

@@ -41,6 +41,8 @@
 import com.oceanbase.odc.service.common.util.WebResponseUtils;
 import com.oceanbase.odc.service.dlm.model.RateLimitConfiguration;
 import com.oceanbase.odc.service.schedule.ScheduleService;
+import com.oceanbase.odc.service.schedule.export.model.ScheduleTerminateCmd;
+import com.oceanbase.odc.service.schedule.export.model.ScheduleTerminateResult;
 import com.oceanbase.odc.service.schedule.model.ChangeScheduleResp;
 import com.oceanbase.odc.service.schedule.model.CreateScheduleReq;
 import com.oceanbase.odc.service.schedule.model.OperationType;
@@ -59,6 +61,8 @@
 import com.oceanbase.odc.service.schedule.model.ScheduleTaskOverview;
 import com.oceanbase.odc.service.schedule.model.ScheduleType;
 import com.oceanbase.odc.service.schedule.model.UpdateScheduleReq;
+import com.oceanbase.odc.service.state.model.StateName;
+import com.oceanbase.odc.service.state.model.StatefulRoute;
 import com.oceanbase.odc.service.task.executor.logger.LogUtils;
 import com.oceanbase.odc.service.task.model.OdcTaskLogLevel;
 
@@ -286,6 +290,23 @@ public SuccessResponse<RateLimitConfiguration> updateLimiterConfig(@PathVariable
         return Responses.single(scheduleService.updateDlmRateLimit(id, limiterConfig));
     }
 
+    @RequestMapping(value = "/schedules/asyncTerminate", method = RequestMethod.POST)
+    public SuccessResponse<String> startTerminateScheduleAndTask(@RequestBody ScheduleTerminateCmd cmd) {
+        return Responses.ok(scheduleService.startTerminateScheduleAndTask(cmd));
+    }
+
+    @RequestMapping(value = "/schedules/asyncTerminateResult", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#terminateId")
+    public SuccessResponse<List<ScheduleTerminateResult>> getTerminateScheduleResult(String terminateId) {
+        return Responses.ok(scheduleService.getTerminateScheduleResult(terminateId));
+    }
+
+    @RequestMapping(value = "/schedules/asyncTerminateLog", method = RequestMethod.GET)
+    @StatefulRoute(stateName = StateName.UUID_STATEFUL_ID, stateIdExpression = "#terminateId")
+    public SuccessResponse<String> getTerminateScheduleLog(String terminateId) {
+        return Responses.ok(scheduleService.getTerminateLog(terminateId));
+    }
+
     @RequestMapping(value = "/schedules/stats", method = RequestMethod.GET)
     public ListResponse<ScheduleStat> getScheduleStats(
             @RequestParam(required = false, name = "types") Set<ScheduleType> types,

server/odc-server/src/main/resources/log4j2.xml

@@ -918,6 +918,10 @@
             <AppenderRef ref="RouteLogRunnableAppender"/>
         </Logger>
 
+        <Logger name="com.oceanbase.odc.service.schedule.ScheduleService" level="INFO" additivity="true">
+            <AppenderRef ref="RouteLogRunnableAppender"/>
+        </Logger>
+
         <!-- ODC程序日志输出，输出级别 INFO -->
         <Logger name="com.oceanbase.odc" level="INFO" additivity="false">
             <AppenderRef ref="OdcFileAppender"/>

server/odc-service/src/main/java/com/oceanbase/odc/config/ScheduleConfiguration.java

@@ -311,8 +311,8 @@ public ThreadPoolTaskExecutor queryProfileMonitorExecutor() {
         return executor;
     }
 
-    @Bean(name = "scheduleImportExecutor")
-    public ThreadPoolTaskExecutor scheduleImportExecutor() {
+    @Bean(name = "commonAsyncTaskExecutor")
+    public ThreadPoolTaskExecutor commonAsyncTaskExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
         int minPoolSize = Math.max(SystemUtils.availableProcessors(), 4);
         executor.setCorePoolSize(minPoolSize);
@@ -324,7 +324,7 @@ public ThreadPoolTaskExecutor scheduleImportExecutor() {
         executor.setTaskDecorator(new TraceDecorator<>());
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
         executor.initialize();
-        log.info("scheduleImportExecutor initialized");
+        log.info("commonAsyncTaskExecutor initialized");
         return executor;
     }
 

server/odc-service/src/main/java/com/oceanbase/odc/service/collaboration/project/ProjectService.java

@@ -85,6 +85,7 @@
 import com.oceanbase.odc.service.iam.ProjectPermissionValidator;
 import com.oceanbase.odc.service.iam.ResourceRoleService;
 import com.oceanbase.odc.service.iam.UserOrganizationService;
+import com.oceanbase.odc.service.iam.UserService;
 import com.oceanbase.odc.service.iam.auth.AuthenticationFacade;
 import com.oceanbase.odc.service.iam.auth.AuthorizationFacade;
 import com.oceanbase.odc.service.iam.model.User;
@@ -164,7 +165,8 @@ public class ProjectService {
     @Autowired
     @Lazy
     private FlowInstanceService flowInstanceService;
-
+    @Autowired
+    private UserService userService;
     @Value("${odc.integration.bastion.enabled:false}")
     private boolean bastionEnabled;
 
@@ -340,7 +342,11 @@ public Page<Project> list(@Valid QueryProjectParams params, @NotNull Pageable pa
 
     @SkipAuthorize("odc internal usage")
     public List<Project> listByIds(@NotEmpty Set<Long> ids) {
-        return repository.findAllById(ids).stream().map(this::entityToModel).collect(Collectors.toList());
+        List<Project> projects =
+                repository.findAllById(ids).stream().map(projectMapper::entityToModel).collect(Collectors.toList());
+        userService.assignInnerUserByCreatorId(projects, c -> c.getCreator().getId(), Project::setCreator);
+        userService.assignInnerUserByCreatorId(projects, c -> c.getLastModifier().getId(), Project::setCreator);
+        return projects;
     }
 
     private Page<ProjectEntity> innerList(@Valid QueryProjectParams params, @NotNull Pageable pageable,

server/odc-service/src/main/java/com/oceanbase/odc/service/common/FutureCache.java

@@ -32,6 +32,7 @@ public class FutureCache {
 
     private final Cache<String, Future<?>> tempId2Future =
             Caffeine.newBuilder().expireAfterWrite(30, TimeUnit.MINUTES)
+                    .maximumSize(1000L)
                     .removalListener((String key, Future<?> future, RemovalCause cause) -> {
                         if (future != null) {
                             future.cancel(true);