FINERACT-1678

- COB Partitioning tasks creates fully balanced id ranges where the number of the loans between the min max are matching with the defined pageSize.

fineract-provider/src/main/java/org/apache/fineract/cob/api/InternalCOBApiResource.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.fineract.cob.api;
+
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.UriInfo;
+import java.time.LocalDate;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.cob.data.LoanCOBPartition;
+import org.apache.fineract.cob.loan.LoanCOBConstant;
+import org.apache.fineract.cob.loan.RetrieveLoanIdService;
+import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
+import org.apache.fineract.infrastructure.core.api.ApiRequestParameterHelper;
+import org.apache.fineract.infrastructure.core.serialization.ApiRequestJsonSerializationSettings;
+import org.apache.fineract.infrastructure.core.serialization.ToApiJsonSerializer;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Profile("test")
+@Component
+@Path("/v1/internal/cob")
+@RequiredArgsConstructor
+@Slf4j
+public class InternalCOBApiResource implements InitializingBean {
+
+    private final RetrieveLoanIdService retrieveLoanIdService;
+    private final ApiRequestParameterHelper apiRequestParameterHelper;
+    private final ToApiJsonSerializer<List> toApiJsonSerializerForList;
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        log.warn("------------------------------------------------------------");
+        log.warn("                                                            ");
+        log.warn("DO NOT USE THIS IN PRODUCTION!");
+        log.warn("Internal client services mode is enabled");
+        log.warn("DO NOT USE THIS IN PRODUCTION!");
+        log.warn("                                                            ");
+        log.warn("------------------------------------------------------------");
+    }
+
+    @GET
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Path("partitions/{partitionSize}")
+    public String getCobPartitions(@Context final UriInfo uriInfo, @PathParam("partitionSize") int partitionSize) {
+        LocalDate businessDate = ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.BUSINESS_DATE);
+        log.info("RetrieveLoanCOBPartitions is called with partitionSize {} for {}", partitionSize, businessDate);
+        List<LoanCOBPartition> loanCOBPartitions = retrieveLoanIdService.retrieveLoanCOBPartitions(LoanCOBConstant.NUMBER_OF_DAYS_BEHIND,
+                businessDate, false, partitionSize);
+        final ApiRequestJsonSerializationSettings settings = this.apiRequestParameterHelper.process(uriInfo.getQueryParameters());
+        return toApiJsonSerializerForList.serialize(settings, loanCOBPartitions);
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java

@@ -20,7 +20,6 @@ package org.apache.fineract.cob.loan;
 
 import org.apache.fineract.cob.COBBusinessStepService;
 import org.apache.fineract.cob.common.CustomJobParameterResolver;
-import org.apache.fineract.cob.data.LoanCOBParameter;
 import org.apache.fineract.cob.listener.COBExecutionListenerRunner;
 import org.apache.fineract.infrastructure.event.business.service.BusinessEventNotifierService;
 import org.apache.fineract.infrastructure.jobs.service.JobName;
@@ -38,7 +37,6 @@ import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -78,19 +76,15 @@ public class LoanCOBManagerConfiguration {
 
     @Bean
     @JobScope
-    public LoanCOBPartitioner partitioner(@Value("#{jobExecutionContext['loanCobParameter']}") LoanCOBParameter loanCOBParameter) {
-        return new LoanCOBPartitioner(propertyService, cobBusinessStepService, jobOperator, jobExplorer, loanCOBParameter);
+    public LoanCOBPartitioner partitioner() {
+        return new LoanCOBPartitioner(propertyService, cobBusinessStepService, retrieveLoanIdService, jobOperator, jobExplorer,
+                LoanCOBConstant.NUMBER_OF_DAYS_BEHIND);
     }
 
     @Bean
     public Step loanCOBStep() {
         return stepBuilderFactory.get(LoanCOBConstant.LOAN_COB_PARTITIONER_STEP)
-                .partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner(null)).outputChannel(outboundRequests).build();
-    }
-
-    @Bean
-    public Step loanIdParameterStep() {
-        return new StepBuilder("Set loan ID parameter - Step", jobRepository).tasklet(loanIdParameterTasklet(), transactionManager).build();
+                .partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner()).outputChannel(outboundRequests).build();
     }
 
     @Bean
@@ -105,12 +99,6 @@ public class LoanCOBManagerConfiguration {
                 .build();
     }
 
-    @Bean
-    @JobScope
-    public LoanIdParameterTasklet loanIdParameterTasklet() {
-        return new LoanIdParameterTasklet(retrieveLoanIdService, customJobParameterResolver);
-    }
-
     @Bean
     @JobScope
     public ResolveLoanCOBCustomJobParametersTasklet resolveCustomJobParametersTasklet() {
@@ -128,7 +116,7 @@ public class LoanCOBManagerConfiguration {
         return new JobBuilder(JobName.LOAN_COB.name(), jobRepository) //
                 .listener(new COBExecutionListenerRunner(applicationContext, JobName.LOAN_COB.name())) //
                 .start(resolveCustomJobParametersStep()) //
-                .next(loanIdParameterStep()).next(loanCOBStep()).next(stayedLockedStep()) //
+                .next(loanCOBStep()).next(stayedLockedStep()) //
                 .incrementer(new RunIdIncrementer()) //
                 .build();
     }
@@ -136,7 +124,7 @@ public class LoanCOBManagerConfiguration {
     @Bean
     public ExecutionContextPromotionListener customJobParametersPromotionListener() {
         ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
-        listener.setKeys(new String[] { LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME });
+        listener.setKeys(new String[] { LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME, LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME });
         return listener;
     }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBPartitioner.java

@@ -18,18 +18,19 @@
  */
 package org.apache.fineract.cob.loan;
 
-import com.google.common.collect.Lists;
-import java.util.HashMap;
+import java.time.LocalDate;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
-import java.util.stream.LongStream;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.cob.COBBusinessStepService;
 import org.apache.fineract.cob.data.BusinessStepNameAndOrder;
 import org.apache.fineract.cob.data.LoanCOBParameter;
+import org.apache.fineract.cob.data.LoanCOBPartition;
 import org.apache.fineract.infrastructure.jobs.service.JobName;
 import org.apache.fineract.infrastructure.springbatch.PropertyService;
 import org.jetbrains.annotations.NotNull;
@@ -40,6 +41,8 @@ import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.launch.NoSuchJobExecutionException;
 import org.springframework.batch.core.partition.support.Partitioner;
 import org.springframework.batch.item.ExecutionContext;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.util.StopWatch;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -49,10 +52,18 @@ public class LoanCOBPartitioner implements Partitioner {
 
     private final PropertyService propertyService;
     private final COBBusinessStepService cobBusinessStepService;
+    private final RetrieveLoanIdService retrieveLoanIdService;
     private final JobOperator jobOperator;
     private final JobExplorer jobExplorer;
 
-    private final LoanCOBParameter minAndMaxLoanId;
+    private final Long numberOfDays;
+
+    @Value("#{jobExecutionContext['BusinessDate']}")
+    @Setter
+    private LocalDate businessDate;
+    @Value("#{jobExecutionContext['IS_CATCH_UP']}")
+    @Setter
+    private Boolean isCatchUp;
 
     @NotNull
     @Override
@@ -64,31 +75,38 @@ public class LoanCOBPartitioner implements Partitioner {
     }
 
     private Map<String, ExecutionContext> getPartitions(int partitionSize, Set<BusinessStepNameAndOrder> cobBusinessSteps) {
-        Map<String, ExecutionContext> partitions = new HashMap<>();
         if (cobBusinessSteps.isEmpty()) {
             stopJobExecution();
             return Map.of();
         }
-        if (!Objects.isNull(minAndMaxLoanId)) {
-            List<Long> loanIdsInRange = LongStream.rangeClosed(minAndMaxLoanId.getMinLoanId(), minAndMaxLoanId.getMaxLoanId()).boxed()
-                    .toList();
-            List<List<Long>> loanIdPartitions = Lists.partition(loanIdsInRange, partitionSize);
-            for (int i = 0; i < loanIdPartitions.size(); i++) {
-                createNewPartition(partitions, i + 1, cobBusinessSteps, loanIdPartitions.get(i));
-            }
-        } else {
-            createNewPartition(partitions, 1, cobBusinessSteps, List.of(0L));
+        StopWatch sw = new StopWatch();
+        sw.start();
+        List<LoanCOBPartition> loanCOBPartitions = new ArrayList<>(retrieveLoanIdService.retrieveLoanCOBPartitions(numberOfDays,
+                businessDate, isCatchUp != null ? isCatchUp : false, partitionSize));
+        sw.stop();
+        // if there is no loan to be closed, we still would like to create at least one partition
+
+        if (loanCOBPartitions.size() == 0) {
+            loanCOBPartitions.add(new LoanCOBPartition(0L, 0L, 1L, 0L));
         }
-        return partitions;
+        log.info(
+                "LoanCOBPartitioner found {} loans to be processed as part of COB. {} partitions were created using partition size {}. RetrieveLoanCOBPartitions was executed in {} ms.",
+                getLoanCount(loanCOBPartitions), loanCOBPartitions.size(), partitionSize, sw.getTotalTimeMillis());
+        return loanCOBPartitions.stream()
+                .collect(Collectors.toMap(l -> PARTITION_PREFIX + l.getPageNo(), l -> createNewPartition(cobBusinessSteps, l)));
     }
 
-    private void createNewPartition(Map<String, ExecutionContext> partitions, int partitionIndex,
-            Set<BusinessStepNameAndOrder> cobBusinessSteps, List<Long> loanIds) {
+    private long getLoanCount(List<LoanCOBPartition> loanCOBPartitions) {
+        return loanCOBPartitions.stream().map(LoanCOBPartition::getCount).reduce(0L, Long::sum);
+    }
+
+    private ExecutionContext createNewPartition(Set<BusinessStepNameAndOrder> cobBusinessSteps, LoanCOBPartition loanCOBPartition) {
         ExecutionContext executionContext = new ExecutionContext();
         executionContext.put(LoanCOBConstant.BUSINESS_STEPS, cobBusinessSteps);
-        executionContext.put(LoanCOBConstant.LOAN_COB_PARAMETER, new LoanCOBParameter(loanIds.get(0), loanIds.get(loanIds.size() - 1)));
-        executionContext.put("partition", PARTITION_PREFIX + partitionIndex);
-        partitions.put(PARTITION_PREFIX + partitionIndex, executionContext);
+        executionContext.put(LoanCOBConstant.LOAN_COB_PARAMETER,
+                new LoanCOBParameter(loanCOBPartition.getMinId(), loanCOBPartition.getMaxId()));
+        executionContext.put("partition", PARTITION_PREFIX + loanCOBPartition.getPageNo());
+        return executionContext;
     }
 
     private void stopJobExecution() {

fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanIdParameterTasklet.java

@@ -1,58 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.apache.fineract.cob.loan;
-
-import java.time.LocalDate;
-import java.util.Objects;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.fineract.cob.common.CustomJobParameterResolver;
-import org.apache.fineract.cob.data.LoanCOBParameter;
-import org.springframework.batch.core.StepContribution;
-import org.springframework.batch.core.scope.context.ChunkContext;
-import org.springframework.batch.core.step.tasklet.Tasklet;
-import org.springframework.batch.repeat.RepeatStatus;
-
-@Slf4j
-@RequiredArgsConstructor
-public class LoanIdParameterTasklet implements Tasklet {
-
-    private final RetrieveLoanIdService retrieveLoanIdService;
-    private final CustomJobParameterResolver customJobParameterResolver;
-
-    @Override
-    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
-        String businessDateParameter = (String) contribution.getStepExecution().getJobExecution().getExecutionContext()
-                .get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME);
-        LocalDate businessDate = LocalDate.parse(Objects.requireNonNull(businessDateParameter));
-        LoanCOBParameter minAndMaxLoanId = retrieveLoanIdService.retrieveMinAndMaxLoanIdsNDaysBehind(LoanCOBConstant.NUMBER_OF_DAYS_BEHIND,
-                businessDate, customJobParameterResolver.getCustomJobParameterById(chunkContext.getStepContext().getStepExecution(),
-                        LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME).map(Boolean::parseBoolean).orElse(false));
-        if (Objects.isNull(minAndMaxLoanId)
-                || (Objects.isNull(minAndMaxLoanId.getMinLoanId()) && Objects.isNull(minAndMaxLoanId.getMaxLoanId()))) {
-            contribution.getStepExecution().getJobExecution().getExecutionContext().put(LoanCOBConstant.LOAN_COB_PARAMETER,
-                    new LoanCOBParameter(0L, 0L));
-            return RepeatStatus.FINISHED;
-        }
-
-        contribution.getStepExecution().getJobExecution().getExecutionContext().put(LoanCOBConstant.LOAN_COB_PARAMETER, minAndMaxLoanId);
-
-        return RepeatStatus.FINISHED;
-    }
-}

fineract-provider/src/main/java/org/apache/fineract/cob/loan/RetrieveAllNonClosedLoanIdServiceImpl.java

@@ -18,28 +18,53 @@
  */
 package org.apache.fineract.cob.loan;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.time.LocalDate;
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.apache.fineract.cob.data.LoanCOBParameter;
+import org.apache.fineract.cob.data.LoanCOBPartition;
 import org.apache.fineract.cob.data.LoanIdAndExternalIdAndAccountNo;
 import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
 import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 
 @RequiredArgsConstructor
 public class RetrieveAllNonClosedLoanIdServiceImpl implements RetrieveLoanIdService {
 
     private final LoanRepository loanRepository;
 
+    private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
+
     @Override
-    public LoanCOBParameter retrieveMinAndMaxLoanIdsNDaysBehind(Long numberOfDays, LocalDate businessDate, boolean isCatchUp) {
+    public List<LoanCOBPartition> retrieveLoanCOBPartitions(Long numberOfDays, LocalDate businessDate, boolean isCatchUp,
+            int partitionSize) {
+        StringBuilder sql = new StringBuilder();
+        sql.append("select min(id) as min, max(id) as max, page, count(id) as count from ");
+        sql.append("  (select floor(((row_number() over(order by id))-1) / :pageSize) as page, t.* from ");
+        sql.append("      (select id from m_loan where loan_status_id in (:statusIds) and ");
         if (isCatchUp) {
-            return loanRepository.findMinAndMaxNonClosedLoanIdsByLastClosedBusinessDateNotNull(businessDate.minusDays(numberOfDays));
+            sql.append("last_closed_business_date = :businessDate ");
         } else {
-            return loanRepository.findMinAndMaxNonClosedLoanIdsByLastClosedBusinessDate(businessDate.minusDays(numberOfDays));
+            sql.append("(last_closed_business_date = :businessDate or last_closed_business_date is null) ");
         }
+        sql.append("order by id) t) t2 ");
+        sql.append("group by page ");
+        sql.append("order by page");
+
+        MapSqlParameterSource parameters = new MapSqlParameterSource();
+        parameters.addValue("pageSize", partitionSize);
+        parameters.addValue("statusIds", List.of(100, 200, 300, 303, 304));
+        parameters.addValue("businessDate", businessDate.minusDays(numberOfDays));
+        return namedParameterJdbcTemplate.query(sql.toString(), parameters, RetrieveAllNonClosedLoanIdServiceImpl::mapRow);
+    }
+
+    private static LoanCOBPartition mapRow(ResultSet rs, int rowNum) throws SQLException {
+        return new LoanCOBPartition(rs.getLong("min"), rs.getLong("max"), rs.getLong("page"), rs.getLong("count"));
     }
 
     @Override

fineract-provider/src/main/java/org/apache/fineract/cob/loan/RetrieveLoanIdConfiguration.java

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 
 @Configuration
 public class RetrieveLoanIdConfiguration {
@@ -30,9 +31,12 @@ public class RetrieveLoanIdConfiguration {
     @Autowired
     private LoanRepository loanRepository;
 
+    @Autowired
+    private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
+
     @Bean
     @ConditionalOnMissingBean
     public RetrieveLoanIdService retrieveLoanIdService() {
-        return new RetrieveAllNonClosedLoanIdServiceImpl(loanRepository);
+        return new RetrieveAllNonClosedLoanIdServiceImpl(loanRepository, namedParameterJdbcTemplate);
     }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/loan/RetrieveLoanIdService.java

@@ -21,13 +21,14 @@ package org.apache.fineract.cob.loan;
 import java.time.LocalDate;
 import java.util.List;
 import org.apache.fineract.cob.data.LoanCOBParameter;
+import org.apache.fineract.cob.data.LoanCOBPartition;
 import org.apache.fineract.cob.data.LoanIdAndExternalIdAndAccountNo;
 import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
 import org.springframework.data.repository.query.Param;
 
 public interface RetrieveLoanIdService {
 
-    LoanCOBParameter retrieveMinAndMaxLoanIdsNDaysBehind(Long numberOfDays, LocalDate businessDate, boolean isCatchUp);
+    List<LoanCOBPartition> retrieveLoanCOBPartitions(Long numberOfDays, LocalDate businessDate, boolean isCatchUp, int partitionSize);
 
     List<LoanIdAndLastClosedBusinessDate> retrieveLoanIdsBehindDateOrNull(LocalDate businessDate, List<Long> loanIds);