CBSNexora/cbsnexora-backend
4
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

FINERACT-1678: Soft-lock Loan COB relevant loan accounts upon startup

Browse Source
This commit is contained in:
Adam Saghy
2022-09-13 16:47:12 +02:00
committed by Arnold Gálovics
parent f258aa258c
commit 44db1d7b94
15 changed files with 325 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fineract-provider/src/main/java/org/apache/fineract/cob/loan/ApplyLoanLockTasklet.java
+1 -1
View File
@@ -67,7 +67,7 @@ public class ApplyLoanLockTasklet implements Tasklet {
accountLockRepository.save(loanAccountLock);
}
executionContext.put(LoanCOBWorkerConfiguration.ALREADY_LOCKED_LOAN_IDS, new ArrayList<>(alreadyUnderProcessingAccountIds));
executionContext.put(LoanCOBConstant.ALREADY_LOCKED_LOAN_IDS, new ArrayList<>(alreadyUnderProcessingAccountIds));
return RepeatStatus.FINISHED;
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/FetchAndLockLoanTasklet.java
+79
View File
@@ -0,0 +1,79 @@
/**
* 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.util.ArrayList;
import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.cob.domain.LoanAccountLock;
import org.apache.fineract.cob.domain.LoanAccountLockRepository;
import org.apache.fineract.cob.domain.LockOwner;
import org.jetbrains.annotations.NotNull;
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 FetchAndLockLoanTasklet implements Tasklet {
private final LoanAccountLockRepository loanAccountLockRepository;
private final RetrieveLoanIdService retrieveLoanIdService;
@Override
public RepeatStatus execute(@NotNull StepContribution contribution, @NotNull ChunkContext chunkContext) throws Exception {
List<Long> allNonClosedLoanIds = retrieveLoanIdService.retrieveLoanIds();
if (allNonClosedLoanIds.isEmpty()) {
return RepeatStatus.FINISHED;
}
List<Long> remainingIds = new ArrayList<>(allNonClosedLoanIds);
List<LoanAccountLock> loanAccountLocks = loanAccountLockRepository.findAllByLoanIdIn(remainingIds);
List<Long> alreadySoftLockedAccounts = loanAccountLocks.stream()
.filter(e -> LockOwner.LOAN_COB_PARTITIONING.equals(e.getLockOwner())).map(LoanAccountLock::getLoanId).toList();
List<Long> alreadyMarkedForInlineCOBLockedAccounts = loanAccountLocks.stream()
.filter(e -> LockOwner.LOAN_INLINE_COB_PROCESSING.equals(e.getLockOwner())).map(LoanAccountLock::getLoanId).toList();
List<Long> alreadyMarkedForChunkProcessingLockedAccounts = loanAccountLocks.stream()
.filter(e -> LockOwner.LOAN_COB_CHUNK_PROCESSING.equals(e.getLockOwner())).map(LoanAccountLock::getLoanId).toList();
// Remove already hard locked accounts
remainingIds.removeAll(alreadyMarkedForChunkProcessingLockedAccounts);
remainingIds.removeAll(alreadyMarkedForInlineCOBLockedAccounts);
List<Long> lockableLoanAccounts = new ArrayList<>(remainingIds);
lockableLoanAccounts.removeAll(alreadySoftLockedAccounts);
applySoftLock(lockableLoanAccounts);
contribution.getStepExecution().getJobExecution().getExecutionContext().put(LoanCOBConstant.LOAN_IDS, remainingIds);
return RepeatStatus.FINISHED;
}
private void applySoftLock(List<Long> alreadySoftLockedAccounts) {
for (Long loanId : alreadySoftLockedAccounts) {
LoanAccountLock loanAccountLock = new LoanAccountLock(loanId, LockOwner.LOAN_COB_PARTITIONING);
loanAccountLockRepository.save(loanAccountLock);
}
}
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBConstant.java
+2
View File
@@ -26,6 +26,8 @@ public final class LoanCOBConstant {
public static final String BUSINESS_STEP_MAP = "businessStepMap";
public static final String LOAN_COB_WORKER_STEP = "loanCOBWorkerStep";
public static final String ALREADY_LOCKED_LOAN_IDS = "alreadyLockedLoanIds";
private LoanCOBConstant() {
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java
+26 -6
View File
@@ -18,19 +18,24 @@
*/
package org.apache.fineract.cob.loan;
import java.util.List;
import org.apache.fineract.cob.COBBusinessStepService;
import org.apache.fineract.cob.domain.LoanAccountLockRepository;
import org.apache.fineract.cob.listener.COBExecutionListenerRunner;
import org.apache.fineract.infrastructure.jobs.service.JobName;
import org.apache.fineract.infrastructure.springbatch.PropertyService;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
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;
@@ -47,6 +52,8 @@ public class LoanCOBManagerConfiguration {
@Autowired
private RemotePartitioningManagerStepBuilderFactory stepBuilderFactory;
@Autowired
private StepBuilderFactory localStepBuilderFactory;
@Autowired
private PropertyService propertyService;
@Autowired
private DirectChannel outboundRequests;
@@ -58,27 +65,40 @@ public class LoanCOBManagerConfiguration {
private JobExplorer jobExplorer;
@Autowired
private ApplicationContext applicationContext;
@Autowired(required = false)
@Autowired
private RetrieveLoanIdService retrieveLoanIdService;
@Autowired
private LoanAccountLockRepository accountLockRepository;
@Bean
public LoanCOBPartitioner partitioner() {
return new LoanCOBPartitioner(propertyService, cobBusinessStepService, jobOperator, jobExplorer, retrieveLoanIdService);
@JobScope
public LoanCOBPartitioner partitioner(@Value("#{jobExecutionContext['loanIds']}") List<Long> loanIds) {
return new LoanCOBPartitioner(propertyService, cobBusinessStepService, jobOperator, jobExplorer, loanIds);
}
@Bean
public Step loanCOBStep() {
return stepBuilderFactory.get(JobName.LOAN_COB.name()).partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner())
return stepBuilderFactory.get("Loan COB partition - Step").partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner(null))
.outputChannel(outboundRequests).build();
}
@Bean
public Step fetchAndLockStep() {
return localStepBuilderFactory.get("Fetch and Lock loan accounts - Step").tasklet(fetchAndLockLoanTasklet()).build();
}
@Bean
@JobScope
public FetchAndLockLoanTasklet fetchAndLockLoanTasklet() {
return new FetchAndLockLoanTasklet(accountLockRepository, retrieveLoanIdService);
}
@Bean(name = "loanCOBJob")
public Job loanCOBJob() {
return jobBuilderFactory.get(JobName.LOAN_COB.name()) //
.listener(new COBExecutionListenerRunner(applicationContext, JobName.LOAN_COB.name())) //
.start(loanCOBStep()) //
.start(fetchAndLockStep()).next(loanCOBStep()) //
.incrementer(new RunIdIncrementer()) //
.build();
}
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBPartitioner.java
+8 -5
View File
@@ -37,6 +37,7 @@ 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.util.CollectionUtils;
@Slf4j
@RequiredArgsConstructor
@@ -48,7 +49,8 @@ public class LoanCOBPartitioner implements Partitioner {
private final COBBusinessStepService cobBusinessStepService;
private final JobOperator jobOperator;
private final JobExplorer jobExplorer;
private final RetrieveLoanIdService retrieveLoanIdService;
private final List<Long> loanIds;
@NotNull
@Override
@@ -65,15 +67,15 @@ public class LoanCOBPartitioner implements Partitioner {
private Map<String, ExecutionContext> getPartitions(int partitionSize, TreeMap<Long, String> cobBusinessStepMap) {
Map<String, ExecutionContext> partitions = new HashMap<>();
List<Long> allNonClosedLoanIds = retrieveLoanIdService.retrieveLoanIds();
if (allNonClosedLoanIds.isEmpty()) {
if (CollectionUtils.isEmpty(loanIds)) {
stopJobExecution();
return Map.of();
}
int partitionIndex = 1;
int remainingSpace = 0;
createNewPartition(partitions, partitionIndex, cobBusinessStepMap);
for (Long allNonClosedLoanId : allNonClosedLoanIds) {
for (Long loanId : loanIds) {
if (remainingSpace == partitionSize) {
partitionIndex++;
createNewPartition(partitions, partitionIndex, cobBusinessStepMap);
@@ -82,7 +84,7 @@ public class LoanCOBPartitioner implements Partitioner {
String key = PARTITION_PREFIX + partitionIndex;
ExecutionContext executionContext = partitions.get(key);
List<Long> data = (List<Long>) executionContext.get(LoanCOBConstant.LOAN_IDS);
data.add(allNonClosedLoanId);
data.add(loanId);
remainingSpace++;
}
return partitions;
@@ -93,6 +95,7 @@ public class LoanCOBPartitioner implements Partitioner {
ExecutionContext executionContext = new ExecutionContext();
executionContext.put(LoanCOBConstant.LOAN_IDS, new ArrayList<Long>());
executionContext.put(LoanCOBConstant.BUSINESS_STEP_MAP, cobBusinessStepMap);
executionContext.put("partition", PARTITION_PREFIX + partitionIndex);
partitions.put(PARTITION_PREFIX + partitionIndex, executionContext);
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java
+17 -12
View File
@@ -28,12 +28,14 @@ import org.apache.fineract.portfolio.loanaccount.domain.Loan;
import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
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.listener.ExecutionContextPromotionListener;
import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
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.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -44,9 +46,11 @@ import org.springframework.transaction.support.TransactionTemplate;
@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
public class LoanCOBWorkerConfiguration {
public static final String ALREADY_LOCKED_LOAN_IDS = "alreadyLockedLoanIds";
@Autowired
private RemotePartitioningWorkerStepBuilderFactory stepBuilderFactory;
@Autowired
private StepBuilderFactory localStepBuilderFactory;
@Autowired
private PropertyService propertyService;
@Autowired
@@ -69,17 +73,20 @@ public class LoanCOBWorkerConfiguration {
@Bean
public Flow flow() {
return new FlowBuilder<Flow>("cobFlow").start(initialisationStep()).next(applyLockStep()).next(loanBusinessStep()).build();
return new FlowBuilder<Flow>("cobFlow").start(initialisationStep(null)).next(applyLockStep(null)).next(loanBusinessStep(null))
.build();
}
@Bean
public Step initialisationStep() {
return stepBuilderFactory.get("Initialisation - Step").inputChannel(inboundRequests).tasklet(initialiseContext()).build();
@StepScope
public Step initialisationStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
return localStepBuilderFactory.get("Initialisation - Step:" + partitionName).tasklet(initialiseContext()).build();
}
@Bean
public Step loanBusinessStep() {
return stepBuilderFactory.get("Loan COB worker - Step").inputChannel(inboundRequests)
@StepScope
public Step loanBusinessStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
return localStepBuilderFactory.get("Loan Business - Step:" + partitionName)
.<Loan, Loan>chunk(propertyService.getChunkSize(JobName.LOAN_COB.name())).reader(cobWorkerItemReader())
.processor(cobWorkerItemProcessor()).writer(cobWorkerItemWriter()).faultTolerant().skip(Exception.class)
.skipLimit(propertyService.getChunkSize(JobName.LOAN_COB.name()) + 1).listener(loanItemListener())
@@ -87,13 +94,12 @@ public class LoanCOBWorkerConfiguration {
}
@Bean
public Step applyLockStep() {
return stepBuilderFactory.get("Apply lock - Step").inputChannel(inboundRequests).tasklet(applyLock()).listener(promotionListener())
.build();
@StepScope
public Step applyLockStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
return localStepBuilderFactory.get("Apply lock - Step:" + partitionName).tasklet(applyLock()).listener(promotionListener()).build();
}
@Bean
@StepScope
public InitialisationTasklet initialiseContext() {
return new InitialisationTasklet(userRepository);
}
@@ -104,7 +110,6 @@ public class LoanCOBWorkerConfiguration {
}
@Bean
@StepScope
public ApplyLoanLockTasklet applyLock() {
return new ApplyLoanLockTasklet(accountLockRepository);
}
@@ -132,7 +137,7 @@ public class LoanCOBWorkerConfiguration {
@Bean
public ExecutionContextPromotionListener promotionListener() {
ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
listener.setKeys(new String[] { ALREADY_LOCKED_LOAN_IDS });
listener.setKeys(new String[] { LoanCOBConstant.ALREADY_LOCKED_LOAN_IDS });
return listener;
}
}
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanItemReader.java
+1 -1
View File
@@ -49,7 +49,7 @@ public class LoanItemReader implements ItemReader<Loan> {
ExecutionContext executionContext = stepExecution.getExecutionContext();
ExecutionContext jobExecutionContext = stepExecution.getJobExecution().getExecutionContext();
List<Long> loanIds = (List<Long>) executionContext.get(LoanCOBConstant.LOAN_IDS);
alreadyLockedAccounts = (List<Long>) jobExecutionContext.get(LoanCOBWorkerConfiguration.ALREADY_LOCKED_LOAN_IDS);
alreadyLockedAccounts = (List<Long>) jobExecutionContext.get(LoanCOBConstant.ALREADY_LOCKED_LOAN_IDS);
remainingData = new ArrayList<>(loanIds);
}