fix(dlm): the data cleaning task scheduling failed after editing the rate limit configuration

server/integration-test/src/test/java/com/oceanbase/odc/metadb/dlm/DlmLimiterConfigRepositoryTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.metadb.dlm;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import com.oceanbase.odc.ServiceTestEnv;
+
+import cn.hutool.core.lang.Assert;
+
+/**
+ * @Author：tinker
+ * @Date: 2024/1/23 15:20
+ * @Descripition:
+ */
+public class DlmLimiterConfigRepositoryTest extends ServiceTestEnv {
+
+    @Autowired
+    private DlmLimiterConfigRepository dlmLimiterConfigRepository;
+
+    @Test
+    public void testFindByOrderIds() {
+        Random rand = new Random();
+        DlmLimiterConfigEntity entity = createEntity(rand.nextLong());
+        DlmLimiterConfigEntity entity2 = createEntity(rand.nextLong());
+        List<Long> ids = new LinkedList<>();
+        ids.add(entity.getOrderId());
+        ids.add(entity2.getOrderId());
+        List<DlmLimiterConfigEntity> byOrderIdIn = dlmLimiterConfigRepository.findByOrderIdIn(ids);
+        Assert.equals(byOrderIdIn.size(), 2);
+    }
+
+    public DlmLimiterConfigEntity createEntity(Long orderId) {
+        DlmLimiterConfigEntity entity = new DlmLimiterConfigEntity();
+        entity.setRowLimit(100);
+        entity.setDataSizeLimit(100L);
+
+        entity.setBatchSize(100);
+        entity.setOrderId(orderId);
+        dlmLimiterConfigRepository.save(entity);
+        return entity;
+    }
+
+}

server/integration-test/src/test/java/com/oceanbase/odc/service/dlm/RateLimitConfigurationServiceTest.java

@@ -37,7 +37,8 @@ public class RateLimitConfigurationServiceTest extends ServiceTestEnv {
     public void findByOrderIdOrElseDefaultConfig() {
         Long orderId = 1L;
         RateLimitConfiguration config = TestRandom.nextObject(RateLimitConfiguration.class);
-        limiterService.createAndBindToOrder(orderId, config);
+        config.setOrderId(orderId);
+        limiterService.create(config);
         RateLimitConfiguration result = limiterService.getByOrderIdOrElseDefaultConfig(orderId);
         Assert.equals(config, result);
         result = limiterService.getByOrderIdOrElseDefaultConfig(2L);

server/odc-service/src/main/java/com/oceanbase/odc/metadb/dlm/DlmLimiterConfigRepository.java

@@ -15,6 +15,8 @@
  */
 package com.oceanbase.odc.metadb.dlm;
 
+import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 
 import com.oceanbase.odc.config.jpa.OdcJpaRepository;
@@ -28,4 +30,6 @@ public interface DlmLimiterConfigRepository extends OdcJpaRepository<DlmLimiterC
 
     Optional<DlmLimiterConfigEntity> findByOrderId(Long orderId);
 
+    List<DlmLimiterConfigEntity> findByOrderIdIn(Collection<Long> orderId);
+
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/DlmLimiterService.java

@@ -15,23 +15,23 @@
  */
 package com.oceanbase.odc.service.dlm;
 
+import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import com.oceanbase.odc.common.json.JsonUtils;
 import com.oceanbase.odc.core.authority.util.SkipAuthorize;
 import com.oceanbase.odc.core.shared.constant.ResourceType;
 import com.oceanbase.odc.core.shared.exception.NotFoundException;
 import com.oceanbase.odc.metadb.dlm.DlmLimiterConfigEntity;
 import com.oceanbase.odc.metadb.dlm.DlmLimiterConfigRepository;
-import com.oceanbase.odc.service.dlm.model.DataArchiveParameters;
 import com.oceanbase.odc.service.dlm.model.RateLimitConfiguration;
 import com.oceanbase.odc.service.schedule.ScheduleService;
-import com.oceanbase.odc.service.schedule.utils.ScheduleTaskUtil;
 
 /**
  * @Author：tinker
@@ -65,10 +65,9 @@ public class DlmLimiterService {
     @Autowired
     private ScheduleService scheduleService;
 
-    public DlmLimiterConfigEntity createAndBindToOrder(Long orderId, RateLimitConfiguration config) {
+    public DlmLimiterConfigEntity create(RateLimitConfiguration config) {
         checkLimiterConfig(config);
         DlmLimiterConfigEntity entity = mapper.modelToEntity(config);
-        entity.setOrderId(orderId);
         return limiterConfigRepository.save(entity);
     }
 
@@ -81,6 +80,11 @@ public RateLimitConfiguration getByOrderIdOrElseDefaultConfig(Long orderId) {
         }
     }
 
+    public List<RateLimitConfiguration> findByOrderIds(Collection<Long> orderIds) {
+        return limiterConfigRepository.findByOrderIdIn(orderIds).stream().map(mapper::entityToModel)
+                .collect(Collectors.toList());
+    }
+
     @Transactional(rollbackFor = Exception.class)
     public RateLimitConfiguration updateByOrderId(Long orderId, RateLimitConfiguration rateLimit) {
         checkLimiterConfig(rateLimit);
@@ -93,10 +97,6 @@ public RateLimitConfiguration updateByOrderId(Long orderId, RateLimitConfigurati
                     rateLimit.getBatchSize() == null ? entity.getBatchSize() : rateLimit.getBatchSize());
             entity.setDataSizeLimit(rateLimit.getDataSizeLimit() == null ? entity.getDataSizeLimit()
                     : rateLimit.getDataSizeLimit());
-            DataArchiveParameters parameters =
-                    ScheduleTaskUtil.getDataArchiveParameters(scheduleService.nullSafeGetById(orderId));
-            parameters.setRateLimit(mapper.entityToModel(entity));
-            scheduleService.updateJobParametersById(orderId, JsonUtils.toJson(parameters));
             return mapper.entityToModel(limiterConfigRepository.save(entity));
         } else {
             throw new NotFoundException(ResourceType.ODC_DLM_LIMITER_CONFIG, "Id", orderId);

server/odc-service/src/main/java/com/oceanbase/odc/service/dlm/model/RateLimitConfiguration.java

@@ -26,6 +26,8 @@
 @Data
 public class RateLimitConfiguration {
 
+    private Long orderId;
+
     private Integer batchSize;
 
     private Integer rowLimit;

server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/factory/ScheduleResponseMapperFactory.java

@@ -36,6 +36,8 @@
 import com.oceanbase.odc.metadb.schedule.ScheduleEntity;
 import com.oceanbase.odc.service.connection.ConnectionService;
 import com.oceanbase.odc.service.connection.model.ConnectionConfig;
+import com.oceanbase.odc.service.dlm.DlmLimiterService;
+import com.oceanbase.odc.service.dlm.model.RateLimitConfiguration;
 import com.oceanbase.odc.service.flow.ApprovalPermissionService;
 import com.oceanbase.odc.service.flow.model.FlowNodeStatus;
 import com.oceanbase.odc.service.schedule.model.ScheduleDetailResp.ScheduleResponseMapper;
@@ -60,6 +62,9 @@ public class ScheduleResponseMapperFactory {
     @Autowired
     private ConnectionService connectionService;
 
+    @Autowired
+    private DlmLimiterService dlmLimiterService;
+
     public ScheduleResponseMapper generate(ScheduleEntity entity) {
         return generate(Collections.singletonList(entity));
     }
@@ -98,10 +103,15 @@ public ScheduleResponseMapper generate(@NonNull Collection<ScheduleEntity> entit
                 .filter(entry -> flowInstanceId2Candidates.get(entry.getValue()) != null).collect(
                         Collectors.toMap(Entry::getKey, entry -> flowInstanceId2Candidates.get(entry.getValue())));
 
+        Map<Long, RateLimitConfiguration> scheduleId2RateLimitConfiguration =
+                dlmLimiterService.findByOrderIds(scheduleIds).stream().collect(
+                        Collectors.toMap(RateLimitConfiguration::getOrderId, o -> o));
+
         return new ScheduleResponseMapper()
                 .withGetUserById(userEntityMap::get)
                 .withGetApproveInstanceIdById(approveInstanceIdMap::get)
                 .withGetDatasourceById(id2Datasource::get)
-                .withGetCandidatesById(scheduleId2Candidates::get);
+                .withGetCandidatesById(scheduleId2Candidates::get)
+                .withGetDLMRateLimitConfigurationById(scheduleId2RateLimitConfiguration::get);
     }
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/flowtask/AbstractDlmJobPreprocessor.java

@@ -163,7 +163,8 @@ protected void initLimiterConfig(Long scheduleId, RateLimitConfiguration limiter
         if (limiterConfig.getBatchSize() == null) {
             limiterConfig.setBatchSize(defaultLimiterConfig.getBatchSize());
         }
-        limiterService.createAndBindToOrder(scheduleId, limiterConfig);
+        limiterConfig.setOrderId(scheduleId);
+        limiterService.create(limiterConfig);
     }
 
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/model/ScheduleDetailResp.java

@@ -34,6 +34,9 @@
 import com.oceanbase.odc.metadb.schedule.ScheduleEntity;
 import com.oceanbase.odc.service.common.model.InnerUser;
 import com.oceanbase.odc.service.connection.model.ConnectionConfig;
+import com.oceanbase.odc.service.dlm.model.DataArchiveParameters;
+import com.oceanbase.odc.service.dlm.model.DataDeleteParameters;
+import com.oceanbase.odc.service.dlm.model.RateLimitConfiguration;
 import com.oceanbase.odc.service.quartz.model.MisfireStrategy;
 import com.oceanbase.odc.service.quartz.util.QuartzCronExpressionUtils;
 
@@ -124,6 +127,9 @@ public static class ScheduleResponseMapper {
 
         private Function<Long, Set<UserEntity>> getCandidatesById = null;
 
+        private Function<Long, RateLimitConfiguration> getDLMRateLimitConfigurationById = null;
+
+
         public ScheduleResponseMapper withGetUserById(@NonNull Function<Long, UserEntity> getUserById) {
             this.getUserById = getUserById;
             return this;
@@ -147,6 +153,12 @@ public ScheduleResponseMapper withGetCandidatesById(
             return this;
         }
 
+        public ScheduleResponseMapper withGetDLMRateLimitConfigurationById(
+                @NonNull Function<Long, RateLimitConfiguration> getCandidatesById) {
+            this.getDLMRateLimitConfigurationById = getCandidatesById;
+            return this;
+        }
+
 
         public ScheduleDetailResp map(@NonNull ScheduleEntity entity) {
 
@@ -163,8 +175,7 @@ public ScheduleDetailResp map(@NonNull ScheduleEntity entity) {
             if (datasource != null) {
                 resp.setDatasource(new InnerConnection(datasource));
             }
-
-            resp.setJobParameters(entity.getJobParametersJson());
+            resp.setJobParameters(getJobParameters(entity));
             resp.setTriggerConfig(entity.getTriggerConfigJson());
             resp.setNextFireTimes(
                     QuartzCronExpressionUtils.getNextFireTimes(JsonUtils.fromJson(entity.getTriggerConfigJson(),
@@ -191,5 +202,31 @@ public ScheduleDetailResp map(@NonNull ScheduleEntity entity) {
             return resp;
         }
 
+        private String getJobParameters(ScheduleEntity entity) {
+            RateLimitConfiguration rateLimitConfig = getDLMRateLimitConfigurationById.apply(entity.getId());
+            if (rateLimitConfig == null) {
+                return entity.getJobParametersJson();
+            }
+            switch (entity.getJobType()) {
+                case DATA_ARCHIVE: {
+                    DataArchiveParameters parameters =
+                            JsonUtils.fromJson(entity.getJobParametersJson(), DataArchiveParameters.class);
+                    parameters.setRateLimit(rateLimitConfig);
+                    return JsonUtils.toJson(parameters);
+                }
+                case DATA_DELETE: {
+                    DataDeleteParameters parameters =
+                            JsonUtils.fromJson(entity.getJobParametersJson(), DataDeleteParameters.class);
+                    parameters.setRateLimit(rateLimitConfig);
+                    return JsonUtils.toJson(parameters);
+
+                }
+                default: {
+                    return entity.getJobParametersJson();
+                }
+            }
+
+        }
+
     }
 }