diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractPartitionJobConfigValidationCondition.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractPartitionJobConfigValidationCondition.java index 6a71e76e7..4f54c0266 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractPartitionJobConfigValidationCondition.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/condition/FineractPartitionJobConfigValidationCondition.java @@ -36,7 +36,7 @@ public class FineractPartitionJobConfigValidationCondition implements Condition "fineract.partitioned-job", FineractProperties.FineractPartitionedJob.class); if (partitionedJobProperties != null) { List invalidConfigs = partitionedJobProperties.getPartitionedJobProperties().stream() - .filter(isAnyConfigBelowOne().or(FineractPartitionJobConfigValidationCondition::isPartitionRatioInvalid)).toList(); + .filter(isAnyConfigBelowOne().or(FineractPartitionJobConfigValidationCondition::invalidMaxPoolSize)).toList(); if (!invalidConfigs.isEmpty()) { for (FineractProperties.PartitionedJobProperty invalidConfig : invalidConfigs) { log.error( @@ -52,10 +52,11 @@ public class FineractPartitionJobConfigValidationCondition implements Condition private static Predicate isAnyConfigBelowOne() { return partitionedJobProperty -> !(partitionedJobProperty.getPartitionSize() > 0 && partitionedJobProperty.getChunkSize() > 0 - && partitionedJobProperty.getThreadCount() > 0); + && partitionedJobProperty.getThreadPoolCorePoolSize() > 0 && partitionedJobProperty.getThreadPoolMaxPoolSize() > 0 + && partitionedJobProperty.getThreadPoolQueueCapacity() > 0); } - private static boolean isPartitionRatioInvalid(FineractProperties.PartitionedJobProperty partitionedJobProperty) { - return partitionedJobProperty.getPartitionSize() < partitionedJobProperty.getThreadCount() * partitionedJobProperty.getChunkSize(); + private static boolean invalidMaxPoolSize(FineractProperties.PartitionedJobProperty partitionedJobProperty) { + return partitionedJobProperty.getThreadPoolMaxPoolSize() < partitionedJobProperty.getThreadPoolCorePoolSize(); } } diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java index bd149d4eb..f37d27326 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java @@ -155,8 +155,11 @@ public class FineractProperties { private String jobName; private Integer chunkSize; private Integer partitionSize; - private Integer threadCount; + private Integer threadPoolCorePoolSize; + private Integer threadPoolMaxPoolSize; + private Integer threadPoolQueueCapacity; private Integer retryLimit; + } @Getter diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/AbstractLoanItemReader.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/AbstractLoanItemReader.java index 40f39a6d3..8fd736fdd 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/AbstractLoanItemReader.java +++ b/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 { protected final LoanRepository loanRepository; @Setter(AccessLevel.PROTECTED) - private List remainingData; - private Long loanId; + private LinkedBlockingQueue 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 diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java new file mode 100644 index 000000000..c6bccf0ed --- /dev/null +++ b/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(); + }; + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/InlineCOBLoanItemReader.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/InlineCOBLoanItemReader.java index deaa9aaee..f1030f01e 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/InlineCOBLoanItemReader.java +++ b/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 loanIds = (List) executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER); - setRemainingData(new ArrayList<>(loanIds)); + setRemainingData(new LinkedBlockingQueue<>(loanIds)); } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java index c9e417c53..46c6c9736 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java +++ b/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("cobFlow").start(initialisationStep(null)).next(applyLockStep(null)).next(loanBusinessStep(null)) + return new FlowBuilder("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) - .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 stepBuilder = new StepBuilder("Loan Business - Step:" + partitionName, jobRepository) + .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 diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanItemReader.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanItemReader.java index 8e5c0a3ec..c236257f7 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanItemReader.java +++ b/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 lockedByCOBChunkProcessingAccountIds = getLoanIdsLockedWithChunkProcessingLock(loanIds); loanIds.retainAll(lockedByCOBChunkProcessingAccountIds); } - setRemainingData(new ArrayList<>(loanIds)); + setRemainingData(new LinkedBlockingQueue<>(loanIds)); } private List getLoanIdsLockedWithChunkProcessingLock(List loanIds) { diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyService.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyService.java index 053b2536b..6e234a29a 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyService.java +++ b/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); } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyServiceImpl.java index d8cc1ba52..1eab3979c 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/PropertyServiceImpl.java +++ b/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 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 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 function) { List jobProperties = fineractProperties.getPartitionedJob() .getPartitionedJobProperties(); return jobProperties.stream() // .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) // .findFirst() // - .map(FineractProperties.PartitionedJobProperty::getRetryLimit) // + .map(function) // .orElse(1); } } diff --git a/fineract-provider/src/main/resources/application.properties b/fineract-provider/src/main/resources/application.properties index 333e094d2..77b59ad1a 100644 --- a/fineract-provider/src/main/resources/application.properties +++ b/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} diff --git a/fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemReaderTest.java b/fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemReaderTest.java new file mode 100644 index 000000000..9fe36b1dc --- /dev/null +++ b/fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemReaderTest.java @@ -0,0 +1,146 @@ +/** + * 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 static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; +import org.apache.fineract.cob.common.CustomJobParameterResolver; +import org.apache.fineract.cob.data.LoanCOBParameter; +import org.apache.fineract.cob.domain.LoanAccountLock; +import org.apache.fineract.cob.domain.LockOwner; +import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant; +import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; +import org.apache.fineract.portfolio.loanaccount.domain.Loan; +import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.ExecutionContext; + +@ExtendWith(MockitoExtension.class) +class LoanItemReaderTest { + + @Mock + private LoanRepository loanRepository; + + @Mock + private RetrieveLoanIdService retrieveLoanIdService; + + @Mock + private CustomJobParameterResolver customJobParameterResolver; + + @Mock + private LoanLockingService loanLockingService; + + @Mock + private StepExecution stepExecution; + + @Mock + private ExecutionContext executionContext; + + @Mock + private Loan loan; + + @Test + public void testLoanItemReaderSimple() throws Exception { + // given + ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "test", "test", "UTC", null)); + LoanItemReader loanItemReader = new LoanItemReader(loanRepository, retrieveLoanIdService, customJobParameterResolver, + loanLockingService); + when(stepExecution.getExecutionContext()).thenReturn(executionContext); + LoanCOBParameter loanCOBParameter = new LoanCOBParameter(1L, 5L); + when(executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER)).thenReturn(loanCOBParameter); + when(retrieveLoanIdService.retrieveAllNonClosedLoansByLastClosedBusinessDateAndMinAndMaxLoanId(loanCOBParameter, false)) + .thenReturn(new ArrayList<>(List.of(1L, 2L, 3L, 4L, 5L))); + List accountLocks = List.of(1L, 2L, 3L, 4L, 5L).stream() + .map(l -> new LoanAccountLock(l, LockOwner.LOAN_COB_CHUNK_PROCESSING, LocalDate.of(2023, 7, 25))).toList(); + when(loanLockingService.findAllByLoanIdInAndLockOwner(List.of(1L, 2L, 3L, 4L, 5L), LockOwner.LOAN_COB_CHUNK_PROCESSING)) + .thenReturn(accountLocks); + when(loanRepository.findById(anyLong())).thenReturn(Optional.of(loan)); + + // when + then + loanItemReader.beforeStep(stepExecution); + for (long i = 1; i <= 5; i++) { + Loan myLoan = loanItemReader.read(); + Assertions.assertEquals(loan, myLoan); + verify(loanRepository, times(1)).findById(i); + } + + Mockito.verifyNoMoreInteractions(loanRepository); + } + + @Test + public void testLoanItemReaderMultiThreadRead() throws Exception { + // given + ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "test", "test", "UTC", null)); + LoanItemReader loanItemReader = new LoanItemReader(loanRepository, retrieveLoanIdService, customJobParameterResolver, + loanLockingService); + when(stepExecution.getExecutionContext()).thenReturn(executionContext); + LoanCOBParameter loanCOBParameter = new LoanCOBParameter(1L, 100L); + when(executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER)).thenReturn(loanCOBParameter); + when(retrieveLoanIdService.retrieveAllNonClosedLoansByLastClosedBusinessDateAndMinAndMaxLoanId(loanCOBParameter, false)) + .thenReturn(new ArrayList<>(IntStream.rangeClosed(1, 100).boxed().map(Long::valueOf).toList())); + List accountLocks = IntStream.rangeClosed(1, 100).boxed().map(Long::valueOf) + .map(l -> new LoanAccountLock(l, LockOwner.LOAN_COB_CHUNK_PROCESSING, LocalDate.of(2023, 7, 25))).toList(); + when(loanLockingService.findAllByLoanIdInAndLockOwner(IntStream.rangeClosed(1, 100).boxed().map(Long::valueOf).toList(), + LockOwner.LOAN_COB_CHUNK_PROCESSING)).thenReturn(accountLocks); + when(loanRepository.findById(anyLong())).thenReturn(Optional.of(loan)); + + // when + then + loanItemReader.beforeStep(stepExecution); + ExecutorService executorService = Executors.newFixedThreadPool(10); + for (int i = 1; i <= 100; i++) { + Future notUsed = executorService.submit(() -> { + try { + Loan myLoan = loanItemReader.read(); + Assertions.assertEquals(loan, myLoan); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + executorService.shutdown(); + boolean b = executorService.awaitTermination(5L, TimeUnit.SECONDS); + Assertions.assertTrue(b, "Executor did not terminate successfully"); + + // verify that this was called 100times, and for each loan it was called exactly once + for (long i = 1; i <= 100; i++) { + verify(loanRepository, times(1)).findById(i); + } + + Mockito.verifyNoMoreInteractions(loanRepository); + } + +} diff --git a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPartitionJobConfigValidationConditionTest.java b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPartitionJobConfigValidationConditionTest.java index 6342a7a21..a2f0413e9 100644 --- a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPartitionJobConfigValidationConditionTest.java +++ b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPartitionJobConfigValidationConditionTest.java @@ -65,16 +65,6 @@ class FineractPartitionJobConfigValidationConditionTest { } } - @Test - public void testApplicationStartup_ShouldApplicationStartupFails_WhenPartitionSizeInvalid() { - try (MockedStatic propertyFactory = Mockito - .mockStatic(ExplicitConfigurationPropertiesFactory.class)) { - propertyFactory.when(() -> ExplicitConfigurationPropertiesFactory.getProperty(context, "fineract.partitioned-job", - FineractProperties.FineractPartitionedJob.class)).thenReturn(getInvalidPartitionSize()); - assertTrue(testObj.matches(context, metadata)); - } - } - @Test public void testApplicationStartup_ShouldApplicationStart_WhenConfigValid() { try (MockedStatic propertyFactory = Mockito @@ -92,20 +82,9 @@ class FineractPartitionJobConfigValidationConditionTest { partitionedJobProperty.setJobName("LOAN_COB"); partitionedJobProperty.setPartitionSize(100); partitionedJobProperty.setChunkSize(10); - partitionedJobProperty.setThreadCount(10); - jobProperties.add(partitionedJobProperty); - partitionedJob.setPartitionedJobProperties(jobProperties); - return partitionedJob; - } - - private FineractProperties.FineractPartitionedJob getInvalidPartitionSize() { - FineractProperties.FineractPartitionedJob partitionedJob = new FineractProperties.FineractPartitionedJob(); - List jobProperties = new ArrayList<>(); - FineractProperties.PartitionedJobProperty partitionedJobProperty = new FineractProperties.PartitionedJobProperty(); - partitionedJobProperty.setJobName("LOAN_COB"); - partitionedJobProperty.setPartitionSize(99); - partitionedJobProperty.setChunkSize(10); - partitionedJobProperty.setThreadCount(10); + partitionedJobProperty.setThreadPoolCorePoolSize(10); + partitionedJobProperty.setThreadPoolMaxPoolSize(20); + partitionedJobProperty.setThreadPoolQueueCapacity(10); jobProperties.add(partitionedJobProperty); partitionedJob.setPartitionedJobProperties(jobProperties); return partitionedJob; @@ -118,7 +97,9 @@ class FineractPartitionJobConfigValidationConditionTest { partitionedJobProperty.setJobName("LOAN_COB"); partitionedJobProperty.setPartitionSize(0); partitionedJobProperty.setChunkSize(1); - partitionedJobProperty.setThreadCount(1); + partitionedJobProperty.setThreadPoolCorePoolSize(1); + partitionedJobProperty.setThreadPoolMaxPoolSize(1); + partitionedJobProperty.setThreadPoolQueueCapacity(1); jobProperties.add(partitionedJobProperty); partitionedJob.setPartitionedJobProperties(jobProperties); return partitionedJob; diff --git a/fineract-provider/src/test/resources/application-test.properties b/fineract-provider/src/test/resources/application-test.properties index 584ff5a36..caa5ff6b4 100644 --- a/fineract-provider/src/test/resources/application-test.properties +++ b/fineract-provider/src/test/resources/application-test.properties @@ -42,7 +42,9 @@ fineract.mode.batch-enabled=true fineract.partitioned-job.partitioned-job-properties[0].job-name=LOAN_COB fineract.partitioned-job.partitioned-job-properties[0].chunk-size=100 fineract.partitioned-job.partitioned-job-properties[0].partition-size=100 -fineract.partitioned-job.partitioned-job-properties[0].thread-count=1 +fineract.partitioned-job.partitioned-job-properties[0].thread-pool-core-pool-size=1 +fineract.partitioned-job.partitioned-job-properties[0].thread-pool-max-pool-size=1 +fineract.partitioned-job.partitioned-job-properties[0].thread-pool-queue-capacity=1 fineract.partitioned-job.partitioned-job-properties[0].retry-limit=5 fineract.remote-job-message-handler.spring-events.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED:true}