FINERACT-1678

- ThreadPoolTaskExecutor task executor is added to COB workers which allows concurrent processing of chunks.

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

@@ -18,7 +18,7 @@
  */
 package org.apache.fineract.cob.loan;
 
-import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
 import lombok.AccessLevel;
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
@@ -40,21 +40,19 @@ public abstract class AbstractLoanItemReader implements ItemReader<Loan> {
     protected final LoanRepository loanRepository;
 
     @Setter(AccessLevel.PROTECTED)
-    private List<Long> remainingData;
-    private Long loanId;
+    private LinkedBlockingQueue<Long> remainingData;
 
     @Override
     public Loan read() throws Exception {
-        try {
-            if (remainingData.size() > 0) {
-                loanId = remainingData.remove(0);
+        final Long loanId = remainingData.poll();
+        if (loanId != null) {
+            try {
                 return loanRepository.findById(loanId).orElseThrow(() -> new LoanNotFoundException(loanId));
+            } catch (Exception e) {
+                throw new LoanReadException(loanId, e);
             }
-        } catch (Exception e) {
-            throw new LoanReadException(loanId, e);
         }
         return null;
-
     }
 
     @AfterStep

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

@@ -0,0 +1,37 @@
+/**
+ * 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 org.apache.fineract.infrastructure.core.domain.FineractContext;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.core.task.TaskDecorator;
+
+public class ContextAwareTaskDecorator implements TaskDecorator {
+
+    @Override
+    public Runnable decorate(@NotNull Runnable runnable) {
+        final FineractContext context = ThreadLocalContextUtil.getContext();
+        return () -> {
+            ThreadLocalContextUtil.init(context);
+            runnable.run();
+        };
+    }
+
+}

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

@@ -18,8 +18,8 @@
  */
 package org.apache.fineract.cob.loan;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
 import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.core.StepExecution;
@@ -37,6 +37,6 @@ public class InlineCOBLoanItemReader extends AbstractLoanItemReader {
     public void beforeStep(@NotNull StepExecution stepExecution) {
         ExecutionContext executionContext = stepExecution.getJobExecution().getExecutionContext();
         List<Long> loanIds = (List<Long>) executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER);
-        setRemainingData(new ArrayList<>(loanIds));
+        setRemainingData(new LinkedBlockingQueue<>(loanIds));
     }
 }

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

@@ -34,6 +34,7 @@ import org.springframework.batch.core.configuration.annotation.StepScope;
 import org.springframework.batch.core.job.builder.FlowBuilder;
 import org.springframework.batch.core.job.flow.Flow;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.SimpleStepBuilder;
 import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -41,7 +42,10 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
 
@@ -86,7 +90,7 @@ public class LoanCOBWorkerConfiguration {
 
     @Bean
     public Flow flow() {
-        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep(null)).next(applyLockStep(null)).next(loanBusinessStep(null))
+        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep(null)).next(applyLockStep(null)).next(loanBusinessStep(null, null))
                 .next(resetContextStep(null)).build();
     }
 
@@ -97,14 +101,43 @@ public class LoanCOBWorkerConfiguration {
                 .build();
     }
 
+    @Bean
+    public TaskExecutor cobTaskExecutor() {
+        if (propertyService.getThreadPoolMaxPoolSize(LoanCOBConstant.JOB_NAME) == 1) {
+            return new SyncTaskExecutor();
+        }
+        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
+        taskExecutor.setThreadNamePrefix("COB-Thread-");
+        taskExecutor.setThreadGroupName("COB-Thread");
+        taskExecutor.setCorePoolSize(propertyService.getThreadPoolCorePoolSize(JobName.LOAN_COB.name()));
+        taskExecutor.setMaxPoolSize(propertyService.getThreadPoolMaxPoolSize(JobName.LOAN_COB.name()));
+        taskExecutor.setQueueCapacity(propertyService.getThreadPoolQueueCapacity(JobName.LOAN_COB.name()));
+        taskExecutor.setAllowCoreThreadTimeOut(true);
+        taskExecutor.setTaskDecorator(new ContextAwareTaskDecorator());
+        return taskExecutor;
+    }
+
     @Bean
     @StepScope
-    public Step loanBusinessStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
-        return new StepBuilder("Loan Business - Step:" + partitionName, jobRepository)
-                .<Loan, Loan>chunk(propertyService.getChunkSize(JobName.LOAN_COB.name()), transactionManager).reader(cobWorkerItemReader())
-                .processor(cobWorkerItemProcessor()).writer(cobWorkerItemWriter()).faultTolerant().retry(Exception.class)
-                .retryLimit(propertyService.getRetryLimit(LoanCOBConstant.JOB_NAME)).skip(Exception.class)
-                .skipLimit(propertyService.getChunkSize(JobName.LOAN_COB.name()) + 1).listener(loanItemListener()).build();
+    public Step loanBusinessStep(@Value("#{stepExecutionContext['partition']}") String partitionName, TaskExecutor cobTaskExecutor) {
+        SimpleStepBuilder<Loan, Loan> stepBuilder = new StepBuilder("Loan Business - Step:" + partitionName, jobRepository)
+                .<Loan, Loan>chunk(propertyService.getChunkSize(JobName.LOAN_COB.name()), transactionManager) //
+                .reader(cobWorkerItemReader()) //
+                .processor(cobWorkerItemProcessor()) //
+                .writer(cobWorkerItemWriter()) //
+                .faultTolerant() //
+                .retry(Exception.class) //
+                .retryLimit(propertyService.getRetryLimit(LoanCOBConstant.JOB_NAME)) //
+                .skip(Exception.class) //
+                .skipLimit(propertyService.getChunkSize(LoanCOBConstant.JOB_NAME) + 1) //
+                .listener(loanItemListener()) //
+                .transactionManager(transactionManager);
+
+        if (propertyService.getThreadPoolMaxPoolSize(LoanCOBConstant.JOB_NAME) > 1) {
+            stepBuilder.taskExecutor(cobTaskExecutor);
+        }
+
+        return stepBuilder.build();
     }
 
     @Bean

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

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.LinkedBlockingQueue;
 import org.apache.fineract.cob.common.CustomJobParameterResolver;
 import org.apache.fineract.cob.data.LoanCOBParameter;
 import org.apache.fineract.cob.domain.LoanAccountLock;
@@ -64,7 +65,7 @@ public class LoanItemReader extends AbstractLoanItemReader {
             List<Long> lockedByCOBChunkProcessingAccountIds = getLoanIdsLockedWithChunkProcessingLock(loanIds);
             loanIds.retainAll(lockedByCOBChunkProcessingAccountIds);
         }
-        setRemainingData(new ArrayList<>(loanIds));
+        setRemainingData(new LinkedBlockingQueue<>(loanIds));
     }
 
     private List<Long> getLoanIdsLockedWithChunkProcessingLock(List<Long> loanIds) {

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyService.java

@@ -25,4 +25,10 @@ public interface PropertyService {
     Integer getChunkSize(String jobName);
 
     Integer getRetryLimit(String jobName);
+
+    Integer getThreadPoolCorePoolSize(String jobName);
+
+    Integer getThreadPoolMaxPoolSize(String jobName);
+
+    Integer getThreadPoolQueueCapacity(String jobName);
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyServiceImpl.java

@@ -19,6 +19,7 @@
 package org.apache.fineract.infrastructure.springbatch;
 
 import java.util.List;
+import java.util.function.Function;
 import lombok.RequiredArgsConstructor;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.springframework.stereotype.Service;
@@ -31,34 +32,41 @@ public class PropertyServiceImpl implements PropertyService {
 
     @Override
     public Integer getPartitionSize(String jobName) {
-        List<FineractProperties.PartitionedJobProperty> jobProperties = fineractProperties.getPartitionedJob()
-                .getPartitionedJobProperties();
-        return jobProperties.stream() //
-                .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
-                .findFirst() //
-                .map(FineractProperties.PartitionedJobProperty::getPartitionSize) //
-                .orElse(1);
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getPartitionSize);
     }
 
     @Override
     public Integer getChunkSize(String jobName) {
-        List<FineractProperties.PartitionedJobProperty> jobProperties = fineractProperties.getPartitionedJob()
-                .getPartitionedJobProperties();
-        return jobProperties.stream() //
-                .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
-                .findFirst() //
-                .map(FineractProperties.PartitionedJobProperty::getChunkSize) //
-                .orElse(1);
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getChunkSize);
     }
 
     @Override
     public Integer getRetryLimit(String jobName) {
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getRetryLimit);
+    }
+
+    @Override
+    public Integer getThreadPoolCorePoolSize(String jobName) {
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getThreadPoolCorePoolSize);
+    }
+
+    @Override
+    public Integer getThreadPoolMaxPoolSize(String jobName) {
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getThreadPoolMaxPoolSize);
+    }
+
+    @Override
+    public Integer getThreadPoolQueueCapacity(String jobName) {
+        return getProperty(jobName, FineractProperties.PartitionedJobProperty::getThreadPoolQueueCapacity);
+    }
+
+    private Integer getProperty(String jobName, Function<? super FineractProperties.PartitionedJobProperty, Integer> function) {
         List<FineractProperties.PartitionedJobProperty> jobProperties = fineractProperties.getPartitionedJob()
                 .getPartitionedJobProperties();
         return jobProperties.stream() //
                 .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
                 .findFirst() //
-                .map(FineractProperties.PartitionedJobProperty::getRetryLimit) //
+                .map(function) //
                 .orElse(1);
     }
 }

fineract-provider/src/main/resources/application.properties

@@ -64,7 +64,9 @@ fineract.job.stuck-retry-threshold=${FINERACT_JOB_STUCK_RETRY_THRESHOLD:5}
 fineract.partitioned-job.partitioned-job-properties[0].job-name=LOAN_COB
 fineract.partitioned-job.partitioned-job-properties[0].chunk-size=${LOAN_COB_CHUNK_SIZE:100}
 fineract.partitioned-job.partitioned-job-properties[0].partition-size=${LOAN_COB_PARTITION_SIZE:100}
-fineract.partitioned-job.partitioned-job-properties[0].thread-count=${LOAN_COB_THREAD_COUNT:1}
+fineract.partitioned-job.partitioned-job-properties[0].thread-pool-core-pool-size=${LOAN_COB_THREAD_POOL_CORE_POOL_SIZE:5}
+fineract.partitioned-job.partitioned-job-properties[0].thread-pool-max-pool-size=${LOAN_COB_THREAD_POOL_MAX_POOL_SIZE:5}
+fineract.partitioned-job.partitioned-job-properties[0].thread-pool-queue-capacity=${LOAN_COB_THREAD_POOL_QUEUE_CAPACITY:20}
 fineract.partitioned-job.partitioned-job-properties[0].retry-limit=${LOAN_COB_RETRY_LIMIT:5}
 
 fineract.remote-job-message-handler.spring-events.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED:true}