FINERACT-1678: Lock under processing loan accounts

fineract-provider/src/main/java/org/apache/fineract/cob/COBBusinessStepServiceImpl.java

@@ -23,8 +23,10 @@ import java.util.List;
 import java.util.Optional;
 import java.util.TreeMap;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.cob.domain.BatchBusinessStep;
 import org.apache.fineract.cob.domain.BatchBusinessStepRepository;
+import org.apache.fineract.cob.exceptions.BusinessStepException;
 import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
 import org.apache.fineract.infrastructure.core.domain.ActionContext;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
@@ -34,6 +36,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Service;
 
 @Service
+@Slf4j
 @RequiredArgsConstructor
 public class COBBusinessStepServiceImpl implements COBBusinessStepService {
 
@@ -43,10 +46,15 @@ public class COBBusinessStepServiceImpl implements COBBusinessStepService {
 
     @Override
     public <T extends COBBusinessStep<S>, S extends AbstractPersistableCustom> S run(TreeMap<Long, String> executionMap, S item) {
+        if (executionMap == null || executionMap.isEmpty()) {
+            throw new BusinessStepException("Execution map is empty! COB Business step execution skipped!");
+        }
         for (String businessStep : executionMap.values()) {
             try {
                 COBBusinessStep<S> businessStepBean = (COBBusinessStep<S>) applicationContext.getBean(businessStep);
                 item = businessStepBean.execute(item);
+            } catch (Exception e) {
+                throw new BusinessStepException("Error happened during business step execution", e);
             } finally {
                 // Fallback to COB action context after each business step
                 ThreadLocalContextUtil.setActionContext(ActionContext.COB);

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

@@ -65,7 +65,7 @@ public class LoanAccountLockApiResource {
     @Operation(summary = "List locked loan accounts", description = "Returns the locked loan IDs")
     @ApiResponses({
             @ApiResponse(responseCode = "200", description = "OK", content = @Content(array = @ArraySchema(schema = @Schema(implementation = LoanAccountLockApiResourceSwagger.GetLoanAccountLockResponse.class)))) })
-    public String retrieveAllConfiguredBusinessStep(@Context final UriInfo uriInfo, @QueryParam("page") String page,
+    public String retrieveLockedAccounts(@Context final UriInfo uriInfo, @QueryParam("page") String page,
             @QueryParam("limit") String limit) {
 
         List<LoanAccountLock> lockedLoanAccounts = loanAccountLockService.getLockedLoanAccountByPage(Integer.parseInt(page),

fineract-provider/src/main/java/org/apache/fineract/cob/common/InitialisationTasklet.java

@@ -16,17 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.loan;
+package org.apache.fineract.cob.common;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.useradministration.domain.AppUser;
 import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
 import org.jetbrains.annotations.NotNull;
-import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.StepContribution;
-import org.springframework.batch.core.StepExecution;
-import org.springframework.batch.core.StepExecutionListener;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.repeat.RepeatStatus;
@@ -36,10 +33,9 @@ import org.springframework.security.core.context.SecurityContextHolder;
 
 @Slf4j
 @RequiredArgsConstructor
-public class InitialisationTasklet implements Tasklet, StepExecutionListener {
+public class InitialisationTasklet implements Tasklet {
 
     private final AppUserRepositoryWrapper userRepository;
-    private StepExecution stepExecution;
 
     @Override
     public RepeatStatus execute(@NotNull StepContribution contribution, @NotNull ChunkContext chunkContext) throws Exception {
@@ -49,14 +45,4 @@ public class InitialisationTasklet implements Tasklet, StepExecutionListener {
         SecurityContextHolder.getContext().setAuthentication(auth);
         return RepeatStatus.FINISHED;
     }
-
-    @Override
-    public void beforeStep(@NotNull StepExecution stepExecution) {
-        this.stepExecution = stepExecution;
-    }
-
-    @Override
-    public ExitStatus afterStep(@NotNull StepExecution stepExecution) {
-        return ExitStatus.COMPLETED;
-    }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/domain/LoanAccountLock.java

@@ -18,7 +18,7 @@
  */
 package org.apache.fineract.cob.domain;
 
-import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
 import javax.persistence.Column;
 import javax.persistence.Entity;
 import javax.persistence.EnumType;
@@ -28,6 +28,7 @@ import javax.persistence.Table;
 import javax.persistence.Version;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import org.apache.fineract.infrastructure.core.service.DateUtils;
 
 @Entity
 @Table(name = "m_loan_account_locks")
@@ -48,11 +49,27 @@ public class LoanAccountLock {
     private LockOwner lockOwner;
 
     @Column(name = "lock_placed_on", nullable = false)
-    private LocalDateTime lockPlacedOn;
+    private OffsetDateTime lockPlacedOn;
 
     @Column(name = "error")
     private String error;
 
     @Column(name = "stacktrace")
     private String stacktrace;
+
+    public LoanAccountLock(Long loanId, LockOwner lockOwner) {
+        this.loanId = loanId;
+        this.lockOwner = lockOwner;
+        this.lockPlacedOn = DateUtils.getOffsetDateTimeOfTenant();
+    }
+
+    public void setError(String errorMessage, String stacktrace) {
+        this.error = errorMessage;
+        this.stacktrace = stacktrace;
+    }
+
+    public void setNewLockOwner(LockOwner newLockOwner) {
+        this.lockOwner = newLockOwner;
+        this.lockPlacedOn = DateUtils.getOffsetDateTimeOfTenant();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/domain/LoanAccountLockRepository.java

@@ -18,7 +18,16 @@
  */
 package org.apache.fineract.cob.domain;
 
+import java.util.List;
+import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 
-public interface LoanAccountLockRepository extends JpaRepository<LoanAccountLock, Long>, JpaSpecificationExecutor<LoanAccountLock> {}
+public interface LoanAccountLockRepository extends JpaRepository<LoanAccountLock, Long>, JpaSpecificationExecutor<LoanAccountLock> {
+
+    Optional<LoanAccountLock> findByLoanIdAndLockOwner(Long loanId, LockOwner lockOwner);
+
+    void deleteByLoanIdInAndLockOwner(List<Long> loanIds, LockOwner lockOwner);
+
+    List<LoanAccountLock> findAllByLoanIdIn(List<Long> loanIds);
+}

fineract-provider/src/main/java/org/apache/fineract/cob/domain/LockOwner.java

@@ -19,5 +19,5 @@
 package org.apache.fineract.cob.domain;
 
 public enum LockOwner {
-    LOAN_COB, LOAN_INLINE_COB;
+    LOAN_COB_PARTITIONING, LOAN_COB_CHUNK_PROCESSING, LOAN_INLINE_COB_PROCESSING;
 }

fineract-provider/src/main/java/org/apache/fineract/cob/exceptions/BusinessStepException.java

@@ -0,0 +1,30 @@
+/**
+ * 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.exceptions;
+
+public class BusinessStepException extends RuntimeException {
+
+    public BusinessStepException(String message) {
+        super(message);
+    }
+
+    public BusinessStepException(String message, Throwable t) {
+        super(message, t);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/exceptions/LoanAccountWasAlreadyLocked.java

@@ -0,0 +1,26 @@
+/**
+ * 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.exceptions;
+
+public class LoanAccountWasAlreadyLocked extends Exception {
+
+    public LoanAccountWasAlreadyLocked(Long loanId) {
+        super(String.format("Loan is in already locked state! loanId: %d", loanId));
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/exceptions/LoanReadException.java

@@ -0,0 +1,33 @@
+/**
+ * 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.exceptions;
+
+public class LoanReadException extends Exception {
+
+    private final Long id;
+
+    public LoanReadException(Long id, Throwable t) {
+        super(String.format("Loan is in already locked state! loanId: %d", id), t);
+        this.id = id;
+    }
+
+    public Long getId() {
+        return id;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/listener/LoanItemListener.java

@@ -0,0 +1,104 @@
+/**
+ * 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.listener;
+
+import static org.springframework.transaction.TransactionDefinition.PROPAGATION_REQUIRES_NEW;
+
+import java.util.List;
+import java.util.Optional;
+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.apache.fineract.cob.exceptions.LoanReadException;
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+import org.apache.fineract.infrastructure.core.serialization.ThrowableSerialization;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.batch.core.annotation.OnProcessError;
+import org.springframework.batch.core.annotation.OnReadError;
+import org.springframework.batch.core.annotation.OnSkipInProcess;
+import org.springframework.batch.core.annotation.OnSkipInRead;
+import org.springframework.batch.core.annotation.OnSkipInWrite;
+import org.springframework.batch.core.annotation.OnWriteError;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Slf4j
+@RequiredArgsConstructor
+public class LoanItemListener {
+
+    private final LoanAccountLockRepository accountLockRepository;
+
+    private final TransactionTemplate transactionTemplate;
+
+    private void updateAccountLockWithError(List<Long> loanIds, String msg, Throwable e) {
+        transactionTemplate.setPropagationBehavior(PROPAGATION_REQUIRES_NEW);
+        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+
+            @Override
+            protected void doInTransactionWithoutResult(@NotNull TransactionStatus status) {
+                for (Long loanId : loanIds) {
+                    Optional<LoanAccountLock> loanAccountLock = accountLockRepository.findByLoanIdAndLockOwner(loanId,
+                            LockOwner.LOAN_COB_CHUNK_PROCESSING);
+                    loanAccountLock.ifPresent(
+                            accountLock -> accountLock.setError(String.format(msg, loanId), ThrowableSerialization.serialize(e)));
+                }
+            }
+        });
+    }
+
+    @OnReadError
+    public void onReadError(Exception e) {
+        LoanReadException ee = (LoanReadException) e;
+        log.warn("Error was triggered during reading of Loan (id={}) due to: {}", ee.getId(), ThrowableSerialization.serialize(e));
+        updateAccountLockWithError(List.of(ee.getId()), "Loan (id: %d) reading is failed", e);
+    }
+
+    @OnProcessError
+    public void onProcessError(@NotNull Loan item, Exception e) {
+        log.warn("Error was triggered during processing of Loan (id={}) due to: {}", item.getId(), ThrowableSerialization.serialize(e));
+        updateAccountLockWithError(List.of(item.getId()), "Loan (id: %d) processing is failed", e);
+    }
+
+    @OnWriteError
+    public void onWriteError(Exception e, @NotNull List<? extends Loan> items) {
+        List<Long> loanIds = items.stream().map(AbstractPersistableCustom::getId).toList();
+        log.warn("Error was triggered during writing of Loans (ids={}) due to: {}", loanIds, ThrowableSerialization.serialize(e));
+
+        updateAccountLockWithError(loanIds, "Loan (id: %d) writing is failed", e);
+    }
+
+    @OnSkipInRead
+    public void onSkipInRead(@NotNull Exception e) {
+        log.warn("Skipping was triggered during read!");
+    }
+
+    @OnSkipInProcess
+    public void onSkipInProcess(@NotNull Loan item, @NotNull Exception e) {
+        log.warn("Skipping was triggered during processing of Loan (id={})", item.getId());
+    }
+
+    @OnSkipInWrite
+    public void onSkipInWrite(@NotNull Loan item, @NotNull Exception e) {
+        log.warn("Skipping was triggered during writing of Loan (id={})", item.getId());
+    }
+}

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

@@ -0,0 +1,85 @@
+/**
+ * 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 java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+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.item.ExecutionContext;
+import org.springframework.batch.repeat.RepeatStatus;
+
+@Slf4j
+@RequiredArgsConstructor
+public class ApplyLoanLockTasklet implements Tasklet {
+
+    private final LoanAccountLockRepository accountLockRepository;
+
+    @Override
+    public RepeatStatus execute(@NotNull StepContribution contribution, @NotNull ChunkContext chunkContext) throws Exception {
+        ExecutionContext executionContext = contribution.getStepExecution().getExecutionContext();
+        List<Long> loanIds = (List<Long>) executionContext.get(LoanCOBConstant.LOAN_IDS);
+        List<Long> remainingLoanIds = new ArrayList<>(loanIds);
+
+        List<LoanAccountLock> accountLocks = accountLockRepository.findAllByLoanIdIn(remainingLoanIds);
+
+        List<Long> alreadyHardLockedAccountIds = accountLocks.stream()
+                .filter(e -> LockOwner.LOAN_COB_CHUNK_PROCESSING.equals(e.getLockOwner())).map(LoanAccountLock::getLoanId).toList();
+
+        List<Long> alreadyUnderProcessingAccountIds = accountLocks.stream()
+                .filter(e -> LockOwner.LOAN_INLINE_COB_PROCESSING.equals(e.getLockOwner())).map(LoanAccountLock::getLoanId).toList();
+
+        Map<Long, LoanAccountLock> alreadySoftLockedAccountsMap = accountLocks.stream()
+                .filter(e -> LockOwner.LOAN_COB_PARTITIONING.equals(e.getLockOwner()))
+                .collect(Collectors.toMap(LoanAccountLock::getLoanId, Function.identity()));
+
+        remainingLoanIds.removeAll(alreadyHardLockedAccountIds);
+        remainingLoanIds.removeAll(alreadyUnderProcessingAccountIds);
+
+        for (Long loanId : remainingLoanIds) {
+            LoanAccountLock loanAccountLock = addLock(loanId, alreadySoftLockedAccountsMap);
+            accountLockRepository.save(loanAccountLock);
+        }
+
+        executionContext.put(LoanCOBWorkerConfiguration.ALREADY_LOCKED_LOAN_IDS, new ArrayList<>(alreadyUnderProcessingAccountIds));
+        return RepeatStatus.FINISHED;
+    }
+
+    private LoanAccountLock addLock(Long loanId, Map<Long, LoanAccountLock> alreadySoftLockedAccountsMap) {
+        LoanAccountLock loanAccountLock;
+        if (alreadySoftLockedAccountsMap.containsKey(loanId)) {
+            // Upgrade lock
+            loanAccountLock = alreadySoftLockedAccountsMap.get(loanId);
+            loanAccountLock.setNewLockOwner(LockOwner.LOAN_COB_CHUNK_PROCESSING);
+        } else {
+            loanAccountLock = new LoanAccountLock(loanId, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+        }
+        return loanAccountLock;
+    }
+}

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

@@ -0,0 +1,32 @@
+/**
+ * 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;
+
+public final class LoanCOBConstant {
+
+    public static final String JOB_NAME = "LOAN_COB";
+    public static final String LOAN_COB_JOB_NAME = "LOAN_CLOSE_OF_BUSINESS";
+    public static final String LOAN_IDS = "loanIds";
+    public static final String BUSINESS_STEP_MAP = "businessStepMap";
+    public static final String LOAN_COB_WORKER_STEP = "loanCOBWorkerStep";
+
+    private LoanCOBConstant() {
+
+    }
+}

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

@@ -19,9 +19,9 @@
 package org.apache.fineract.cob.loan;
 
 import org.apache.fineract.cob.COBBusinessStepService;
-import org.apache.fineract.cob.COBPropertyService;
 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;
@@ -47,7 +47,7 @@ public class LoanCOBManagerConfiguration {
     @Autowired
     private RemotePartitioningManagerStepBuilderFactory stepBuilderFactory;
     @Autowired
-    private COBPropertyService cobPropertyService;
+    private PropertyService propertyService;
     @Autowired
     private DirectChannel outboundRequests;
     @Autowired
@@ -63,13 +63,13 @@ public class LoanCOBManagerConfiguration {
 
     @Bean
     public LoanCOBPartitioner partitioner() {
-        return new LoanCOBPartitioner(cobPropertyService, cobBusinessStepService, jobOperator, jobExplorer, retrieveLoanIdService);
+        return new LoanCOBPartitioner(propertyService, cobBusinessStepService, jobOperator, jobExplorer, retrieveLoanIdService);
     }
 
     @Bean
     public Step loanCOBStep() {
-        return stepBuilderFactory.get(JobName.LOAN_COB.name()).partitioner("Loan COB worker", partitioner()).outputChannel(outboundRequests)
-                .build();
+        return stepBuilderFactory.get(JobName.LOAN_COB.name()).partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner())
+                .outputChannel(outboundRequests).build();
     }
 
     @Bean(name = "loanCOBJob")

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

@@ -27,9 +27,9 @@ import java.util.TreeMap;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.cob.COBBusinessStepService;
-import org.apache.fineract.cob.COBPropertyService;
 import org.apache.fineract.infrastructure.jobs.service.JobName;
-import org.jetbrains.annotations.Nullable;
+import org.apache.fineract.infrastructure.springbatch.PropertyService;
+import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.launch.JobExecutionNotRunningException;
@@ -42,44 +42,48 @@ import org.springframework.batch.item.ExecutionContext;
 @RequiredArgsConstructor
 public class LoanCOBPartitioner implements Partitioner {
 
-    private final COBPropertyService cobPropertyService;
+    public static final String PARTITION_PREFIX = "partition_";
+
+    private final PropertyService propertyService;
     private final COBBusinessStepService cobBusinessStepService;
     private final JobOperator jobOperator;
     private final JobExplorer jobExplorer;
     private final RetrieveLoanIdService retrieveLoanIdService;
 
-    private static final String PARTITION_PREFIX = "partition_";
-    private static final String JOB_NAME = "LOAN_COB";
-    private static final String LOAN_COB_JOB_NAME = "LOAN_CLOSE_OF_BUSINESS";
-
+    @NotNull
     @Override
     public Map<String, ExecutionContext> partition(int gridSize) {
-        int partitionSize = cobPropertyService.getPartitionSize(JOB_NAME);
+        int partitionSize = propertyService.getPartitionSize(LoanCOBConstant.JOB_NAME);
         TreeMap<Long, String> cobBusinessStepMap = cobBusinessStepService.getCOBBusinessStepMap(LoanCOBBusinessStep.class,
-                LOAN_COB_JOB_NAME);
+                LoanCOBConstant.LOAN_COB_JOB_NAME);
         if (cobBusinessStepMap.isEmpty()) {
-            return stopJobExecution();
+            stopJobExecution();
+            return Map.of();
         }
         return getPartitions(partitionSize, cobBusinessStepMap);
     }
 
     private Map<String, ExecutionContext> getPartitions(int partitionSize, TreeMap<Long, String> cobBusinessStepMap) {
         Map<String, ExecutionContext> partitions = new HashMap<>();
-        List<Integer> allNonClosedLoanIds = retrieveLoanIdService.retrieveLoanIds();
+        List<Long> allNonClosedLoanIds = retrieveLoanIdService.retrieveLoanIds();
         if (allNonClosedLoanIds.isEmpty()) {
-            return stopJobExecution();
+            stopJobExecution();
+            return Map.of();
         }
-        int partitionIndex = 0;
+        int partitionIndex = 1;
+        int remainingSpace = 0;
         createNewPartition(partitions, partitionIndex, cobBusinessStepMap);
-        for (Integer allNonClosedLoanId : allNonClosedLoanIds) {
-            if (partitions.get(PARTITION_PREFIX + partitionIndex).size() == partitionSize) {
+        for (Long allNonClosedLoanId : allNonClosedLoanIds) {
+            if (remainingSpace == partitionSize) {
                 partitionIndex++;
                 createNewPartition(partitions, partitionIndex, cobBusinessStepMap);
+                remainingSpace = 0;
             }
             String key = PARTITION_PREFIX + partitionIndex;
             ExecutionContext executionContext = partitions.get(key);
-            List<Integer> data = (List<Integer>) executionContext.get("loanIds");
+            List<Long> data = (List<Long>) executionContext.get(LoanCOBConstant.LOAN_IDS);
             data.add(allNonClosedLoanId);
+            remainingSpace++;
         }
         return partitions;
     }
@@ -87,13 +91,12 @@ public class LoanCOBPartitioner implements Partitioner {
     private void createNewPartition(Map<String, ExecutionContext> partitions, int partitionIndex,
             TreeMap<Long, String> cobBusinessStepMap) {
         ExecutionContext executionContext = new ExecutionContext();
-        executionContext.put("loanIds", new ArrayList<Integer>());
-        executionContext.put("BusinessStepMap", cobBusinessStepMap);
+        executionContext.put(LoanCOBConstant.LOAN_IDS, new ArrayList<Long>());
+        executionContext.put(LoanCOBConstant.BUSINESS_STEP_MAP, cobBusinessStepMap);
         partitions.put(PARTITION_PREFIX + partitionIndex, executionContext);
     }
 
-    @Nullable
-    private Map<String, ExecutionContext> stopJobExecution() {
+    private void stopJobExecution() {
         Set<JobExecution> runningJobExecutions = jobExplorer.findRunningJobExecutions(JobName.LOAN_COB.name());
         for (JobExecution jobExecution : runningJobExecutions) {
             try {
@@ -103,6 +106,5 @@ public class LoanCOBPartitioner implements Partitioner {
                 throw new RuntimeException(e);
             }
         }
-        return null;
     }
 }

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

@@ -18,42 +18,37 @@
  */
 package org.apache.fineract.cob.loan;
 
-import java.util.ArrayList;
-import java.util.List;
 import org.apache.fineract.cob.COBBusinessStepService;
-import org.apache.fineract.cob.COBPropertyService;
+import org.apache.fineract.cob.common.InitialisationTasklet;
+import org.apache.fineract.cob.domain.LoanAccountLockRepository;
+import org.apache.fineract.cob.listener.LoanItemListener;
 import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.infrastructure.springbatch.PropertyService;
 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.Job;
 import org.springframework.batch.core.Step;
-import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
 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.launch.support.RunIdIncrementer;
+import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
 import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemReader;
-import org.springframework.batch.item.data.RepositoryItemWriter;
 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;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.transaction.support.TransactionTemplate;
 
 @Configuration
 @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
 public class LoanCOBWorkerConfiguration {
 
-    @Autowired
-    private JobBuilderFactory jobBuilderFactory;
+    public static final String ALREADY_LOCKED_LOAN_IDS = "alreadyLockedLoanIds";
     @Autowired
     private RemotePartitioningWorkerStepBuilderFactory stepBuilderFactory;
     @Autowired
-    private COBPropertyService cobPropertyService;
+    private PropertyService propertyService;
     @Autowired
     private LoanRepository loanRepository;
     @Autowired
@@ -61,33 +56,40 @@ public class LoanCOBWorkerConfiguration {
     @Autowired
     private COBBusinessStepService cobBusinessStepService;
     @Autowired
+    private LoanAccountLockRepository accountLockRepository;
+    @Autowired
     private AppUserRepositoryWrapper userRepository;
+    @Autowired
+    private TransactionTemplate transactionTemplate;
 
-    @Bean
-    public Step loanBusinessStep() {
-        return stepBuilderFactory.get("Loan COB worker").inputChannel(inboundRequests)
-                .<Loan, Loan>chunk(cobPropertyService.getChunkSize(JobName.LOAN_COB.name())).reader(itemReader(null))
-                .processor(itemProcessor()).writer(itemWriter()).build();
-    }
-
-    @Bean(name = "Loan COB worker")
+    @Bean(name = LoanCOBConstant.LOAN_COB_WORKER_STEP)
     public Step loanCOBWorkerStep() {
         return stepBuilderFactory.get("Loan COB worker - Step").inputChannel(inboundRequests).flow(flow()).build();
     }
 
-    @Bean
-    public Job loanCOBWorkerJob() {
-        return jobBuilderFactory.get("Loan COB worker").start(loanCOBWorkerStep()).incrementer(new RunIdIncrementer()).build();
-    }
-
     @Bean
     public Flow flow() {
-        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep()).next(loanBusinessStep()).build();
+        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep()).next(applyLockStep()).next(loanBusinessStep()).build();
     }
 
     @Bean
     public Step initialisationStep() {
-        return stepBuilderFactory.get("Initalisation - Step").inputChannel(inboundRequests).tasklet(initialiseContext()).build();
+        return stepBuilderFactory.get("Initialisation - Step").inputChannel(inboundRequests).tasklet(initialiseContext()).build();
+    }
+
+    @Bean
+    public Step loanBusinessStep() {
+        return stepBuilderFactory.get("Loan COB worker - Step").inputChannel(inboundRequests)
+                .<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())
+                .listener(promotionListener()).build();
+    }
+
+    @Bean
+    public Step applyLockStep() {
+        return stepBuilderFactory.get("Apply lock - Step").inputChannel(inboundRequests).tasklet(applyLock()).listener(promotionListener())
+                .build();
     }
 
     @Bean
@@ -97,27 +99,40 @@ public class LoanCOBWorkerConfiguration {
     }
 
     @Bean
-    @StepScope
-    public ItemReader<Loan> itemReader(@Value("#{stepExecutionContext['loanIds']}") List<Integer> data) {
-        List<Integer> remainingData = new ArrayList<>(data);
-        return () -> {
-            if (remainingData.size() > 0) {
-                return loanRepository.findById(remainingData.remove(0).longValue()).orElse(null);
-            }
-            return null;
-        };
+    public LoanItemListener loanItemListener() {
+        return new LoanItemListener(accountLockRepository, transactionTemplate);
     }
 
     @Bean
-    public ItemProcessor<Loan, Loan> itemProcessor() {
+    @StepScope
+    public ApplyLoanLockTasklet applyLock() {
+        return new ApplyLoanLockTasklet(accountLockRepository);
+    }
+
+    @Bean
+    @StepScope
+    public LoanItemReader cobWorkerItemReader() {
+        return new LoanItemReader(loanRepository);
+    }
+
+    @Bean
+    @StepScope
+    public LoanItemProcessor cobWorkerItemProcessor() {
         return new LoanItemProcessor(cobBusinessStepService);
     }
 
     @Bean
-    public RepositoryItemWriter<Loan> itemWriter() {
-        RepositoryItemWriter<Loan> writer = new RepositoryItemWriter<>();
-        writer.setRepository(loanRepository);
-        writer.setMethodName("save");
-        return writer;
+    @StepScope
+    public LoanItemWriter cobWorkerItemWriter() {
+        LoanItemWriter repositoryItemWriter = new LoanItemWriter(accountLockRepository);
+        repositoryItemWriter.setRepository(loanRepository);
+        return repositoryItemWriter;
+    }
+
+    @Bean
+    public ExecutionContextPromotionListener promotionListener() {
+        ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
+        listener.setKeys(new String[] { ALREADY_LOCKED_LOAN_IDS });
+        return listener;
     }
 }

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

@@ -20,36 +20,40 @@ package org.apache.fineract.cob.loan;
 
 import java.util.TreeMap;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.cob.COBBusinessStepService;
 import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.StepExecution;
-import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.annotation.AfterStep;
+import org.springframework.batch.core.annotation.BeforeStep;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemProcessor;
-import org.springframework.stereotype.Component;
 
-@Component
 @RequiredArgsConstructor
-public class LoanItemProcessor implements ItemProcessor<Loan, Loan>, StepExecutionListener {
+@Slf4j
+public class LoanItemProcessor implements ItemProcessor<Loan, Loan> {
 
     private final COBBusinessStepService cobBusinessStepService;
     private StepExecution stepExecution;
 
-    @Override
-    public void beforeStep(StepExecution stepExecution) {
+    @BeforeStep
+    public void beforeStep(@NotNull StepExecution stepExecution) {
         this.stepExecution = stepExecution;
     }
 
     @Override
-    public Loan process(Loan item) throws Exception {
+    public Loan process(@NotNull Loan item) throws Exception {
         ExecutionContext executionContext = stepExecution.getExecutionContext();
-        TreeMap<Long, String> businessStepMap = (TreeMap<Long, String>) executionContext.get("BusinessStepMap");
+        TreeMap<Long, String> businessStepMap = (TreeMap<Long, String>) executionContext.get(LoanCOBConstant.BUSINESS_STEP_MAP);
+
         return cobBusinessStepService.run(businessStepMap, item);
     }
 
-    @Override
-    public ExitStatus afterStep(StepExecution stepExecution) {
+    @AfterStep
+    public ExitStatus afterStep(@NotNull StepExecution stepExecution) {
         return ExitStatus.COMPLETED;
     }
+
 }

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

@@ -0,0 +1,78 @@
+/**
+ * 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.exceptions.LoanAccountWasAlreadyLocked;
+import org.apache.fineract.cob.exceptions.LoanReadException;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.apache.fineract.portfolio.loanaccount.exception.LoanNotFoundException;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.annotation.AfterStep;
+import org.springframework.batch.core.annotation.BeforeStep;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+
+@Slf4j
+@RequiredArgsConstructor
+public class LoanItemReader implements ItemReader<Loan> {
+
+    private final LoanRepository loanRepository;
+    private List<Long> alreadyLockedAccounts;
+    private List<Long> remainingData;
+    private Long loanId;
+
+    @BeforeStep
+    public void beforeStep(@NotNull StepExecution stepExecution) {
+        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);
+        remainingData = new ArrayList<>(loanIds);
+    }
+
+    @Override
+    public Loan read() throws Exception {
+        try {
+            if (remainingData.size() > 0) {
+                loanId = remainingData.remove(0);
+                if (alreadyLockedAccounts != null && alreadyLockedAccounts.remove(loanId)) {
+                    throw new LoanAccountWasAlreadyLocked(loanId);
+                }
+
+                return loanRepository.findById(loanId).orElseThrow(() -> new LoanNotFoundException(loanId));
+            }
+        } catch (Exception e) {
+            throw new LoanReadException(loanId, e);
+        }
+        return null;
+
+    }
+
+    @AfterStep
+    public ExitStatus afterStep(@NotNull StepExecution stepExecution) {
+        return ExitStatus.COMPLETED;
+    }
+}

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

@@ -0,0 +1,45 @@
+/**
+ * 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.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.cob.domain.LoanAccountLockRepository;
+import org.apache.fineract.cob.domain.LockOwner;
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.batch.item.data.RepositoryItemWriter;
+
+@Slf4j
+@RequiredArgsConstructor
+public class LoanItemWriter extends RepositoryItemWriter<Loan> {
+
+    private final LoanAccountLockRepository accountLockRepository;
+
+    @Override
+    public void write(@NotNull List<? extends Loan> items) throws Exception {
+        super.write(items);
+        List<Long> loanIds = items.stream().map(AbstractPersistableCustom::getId).toList();
+
+        accountLockRepository.deleteByLoanIdInAndLockOwner(loanIds, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+    }
+
+}

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

@@ -28,7 +28,7 @@ public class RetrieveAllNonClosedLoanIdServiceImpl implements RetrieveLoanIdServ
     private final LoanRepository loanRepository;
 
     @Override
-    public List<Integer> retrieveLoanIds() {
+    public List<Long> retrieveLoanIds() {
         return loanRepository.findAllNonClosedLoanIds();
     }
 }

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

@@ -22,5 +22,5 @@ import java.util.List;
 
 public interface RetrieveLoanIdService {
 
-    List<Integer> retrieveLoanIds();
+    List<Long> retrieveLoanIds();
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/serialization/ThrowableSerialization.java

@@ -0,0 +1,38 @@
+/**
+ * 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.infrastructure.core.serialization;
+
+import com.google.common.io.CharStreams;
+import java.io.PrintWriter;
+import java.io.Writer;
+
+public final class ThrowableSerialization {
+
+    private ThrowableSerialization() {
+
+    }
+
+    @SuppressWarnings("RegexpSinglelineJava")
+    public static String serialize(Throwable e) {
+        StringBuilder sb = new StringBuilder();
+        Writer w = CharStreams.asWriter(sb);
+        e.printStackTrace(new PrintWriter(w, true));
+        return sb.toString();
+    }
+}

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import java.io.Serializable;
 import lombok.Data;
@@ -24,7 +24,7 @@ import org.apache.fineract.infrastructure.core.domain.FineractContext;
 import org.springframework.batch.integration.partition.StepExecutionRequest;
 
 @Data
-public class COBMessage implements Serializable {
+public class ContextualMessage implements Serializable {
 
     private StepExecutionRequest stepExecutionRequest;
     private FineractContext context;

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

@@ -16,21 +16,22 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import org.apache.fineract.infrastructure.core.domain.ActionContext;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.jetbrains.annotations.NotNull;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.ExecutorChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
 
-public class COBInputChannelInterceptor implements ExecutorChannelInterceptor {
+public class InputChannelInterceptor implements ExecutorChannelInterceptor {
 
     @Override
-    public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
-        COBMessage castedMessage = COBMessage.class.cast(message.getPayload());
+    public Message<?> beforeHandle(Message<?> message, @NotNull MessageChannel channel, @NotNull MessageHandler handler) {
+        ContextualMessage castedMessage = (ContextualMessage) message.getPayload();
         ThreadLocalContextUtil.init(castedMessage.getContext());
         ThreadLocalContextUtil.setActionContext(ActionContext.COB);
         return new GenericMessage<>(castedMessage.getStepExecutionRequest());

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -27,7 +27,7 @@ import org.springframework.integration.channel.DirectChannel;
 @Configuration
 @EnableBatchIntegration
 @ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
-public class COBManagerConfig {
+public class ManagerConfig {
 
     @Bean
     public DirectChannel outboundRequests() {
@@ -35,7 +35,7 @@ public class COBManagerConfig {
     }
 
     @Bean
-    public COBOutputChannelInterceptor outputInterceptor() {
-        return new COBOutputChannelInterceptor();
+    public OutputChannelInterceptor outputInterceptor() {
+        return new OutputChannelInterceptor();
     }
 }

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

@@ -16,23 +16,24 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.integration.async.StepExecutionInterceptor;
 import org.springframework.batch.integration.partition.StepExecutionRequest;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
 
-public class COBOutputChannelInterceptor extends StepExecutionInterceptor {
+public class OutputChannelInterceptor extends StepExecutionInterceptor {
 
     @Override
-    public Message<?> preSend(Message<?> message, MessageChannel channel) {
-        StepExecutionRequest stepExecutionRequest = StepExecutionRequest.class.cast(message.getPayload());
-        COBMessage cobMessage = new COBMessage();
-        cobMessage.setStepExecutionRequest(stepExecutionRequest);
-        cobMessage.setContext(ThreadLocalContextUtil.getContext());
-        return new GenericMessage<>(cobMessage);
+    public Message<?> preSend(Message<?> message, @NotNull MessageChannel channel) {
+        StepExecutionRequest stepExecutionRequest = (StepExecutionRequest) message.getPayload();
+        ContextualMessage contextualMessage = new ContextualMessage();
+        contextualMessage.setStepExecutionRequest(stepExecutionRequest);
+        contextualMessage.setContext(ThreadLocalContextUtil.getContext());
+        return new GenericMessage<>(contextualMessage);
     }
 }

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

@@ -16,9 +16,9 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
-public interface COBPropertyService {
+public interface PropertyService {
 
     Integer getPartitionSize(String jobName);
 

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import java.util.List;
 import lombok.RequiredArgsConstructor;
@@ -25,7 +25,7 @@ import org.springframework.stereotype.Service;
 
 @Service
 @RequiredArgsConstructor
-public class COBPropertyServiceImpl implements COBPropertyService {
+public class PropertyServiceImpl implements PropertyService {
 
     private final FineractProperties fineractProperties;
 
@@ -37,7 +37,7 @@ public class COBPropertyServiceImpl implements COBPropertyService {
                 .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
                 .findFirst() //
                 .map(FineractProperties.PartitionedJobProperty::getPartitionSize) //
-                .orElse(0);
+                .orElse(1);
     }
 
     @Override
@@ -48,6 +48,6 @@ public class COBPropertyServiceImpl implements COBPropertyService {
                 .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
                 .findFirst() //
                 .map(FineractProperties.PartitionedJobProperty::getChunkSize) //
-                .orElse(0);
+                .orElse(1);
     }
 }

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob;
+package org.apache.fineract.infrastructure.springbatch;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
@@ -25,7 +25,7 @@ import org.springframework.integration.channel.QueueChannel;
 
 @Configuration
 @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
-public class COBWorkerConfig {
+public class WorkerConfig {
 
     @Bean
     public QueueChannel inboundRequests() {
@@ -33,7 +33,7 @@ public class COBWorkerConfig {
     }
 
     @Bean
-    public COBInputChannelInterceptor inputInterceptor() {
-        return new COBInputChannelInterceptor();
+    public InputChannelInterceptor inputInterceptor() {
+        return new InputChannelInterceptor();
     }
 }

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;

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

@@ -16,12 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.fineract.cob.COBOutputChannelInterceptor;
-import org.apache.fineract.cob.messagehandler.conditions.JmsManagerCondition;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsManagerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -38,12 +38,12 @@ import org.springframework.integration.jms.dsl.Jms;
 @EnableBatchIntegration
 @Conditional(JmsManagerCondition.class)
 @Import(value = { JmsBrokerConfiguration.class })
-public class COBJmsManagerConfig {
+public class JmsManagerConfig {
 
     @Autowired
     private DirectChannel outboundRequests;
     @Autowired
-    private COBOutputChannelInterceptor outputInterceptor;
+    private OutputChannelInterceptor outputInterceptor;
     @Autowired
     private FineractProperties fineractProperties;
 

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

@@ -16,12 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.fineract.cob.COBInputChannelInterceptor;
-import org.apache.fineract.cob.messagehandler.conditions.JmsWorkerCondition;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -38,12 +38,12 @@ import org.springframework.integration.jms.dsl.Jms;
 @EnableBatchIntegration
 @Conditional(JmsWorkerCondition.class)
 @Import(value = { JmsBrokerConfiguration.class })
-public class COBJmsWorkerConfig {
+public class JmsWorkerConfig {
 
     @Autowired
     private QueueChannel inboundRequests;
     @Autowired
-    private COBInputChannelInterceptor inputInterceptor;
+    private InputChannelInterceptor inputInterceptor;
     @Autowired
     private FineractProperties fineractProperties;
 

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

@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
-import org.apache.fineract.cob.COBOutputChannelInterceptor;
-import org.apache.fineract.cob.messagehandler.conditions.SpringEventManagerCondition;
+import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventManagerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -34,12 +34,12 @@ import org.springframework.integration.handler.LoggingHandler;
 @Configuration
 @EnableBatchIntegration
 @Conditional(SpringEventManagerCondition.class)
-public class COBSpringEventManagerConfig {
+public class SpringEventManagerConfig {
 
     @Autowired
     private DirectChannel outboundRequests;
     @Autowired
-    private COBOutputChannelInterceptor outputInterceptor;
+    private OutputChannelInterceptor outputInterceptor;
 
     @Bean
     public IntegrationFlow outboundFlow() {

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

@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
-import org.apache.fineract.cob.COBInputChannelInterceptor;
-import org.apache.fineract.cob.messagehandler.conditions.SpringEventWorkerCondition;
+import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventWorkerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -35,12 +35,12 @@ import org.springframework.integration.handler.LoggingHandler;
 @Configuration
 @EnableBatchIntegration
 @Conditional(SpringEventWorkerCondition.class)
-public class COBSpringEventWorkerConfig {
+public class SpringEventWorkerConfig {
 
     @Autowired
     private QueueChannel inboundRequests;
     @Autowired
-    private COBInputChannelInterceptor inputInterceptor;
+    private InputChannelInterceptor inputInterceptor;
 
     @Bean
     public IntegrationFlow inboundFlow() {

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsManagerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsWorkerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventManagerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventWorkerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.cob.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java

@@ -166,6 +166,6 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
     boolean existsByExternalId(@Param("externalId") String externalId);
 
     @Query(FIND_ALL_NON_CLOSED)
-    List<Integer> findAllNonClosedLoanIds();
+    List<Long> findAllNonClosedLoanIds();
 
 }

fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml

@@ -67,4 +67,5 @@
     <include file="parts/0045_external_event_table_data_binary.xml" relativeToChangelogFile="true"/>
     <include file="parts/0046_external_event_table_schema_info.xml" relativeToChangelogFile="true"/>
     <include file="parts/0047_add_loan_delinquency_tags_business_step.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0048_rework_loan_account_locks.xml" relativeToChangelogFile="true"/>
 </databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0048_rework_loan_account_locks.xml

@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
+
+    <changeSet id="1" author="fineract">
+        <dropColumn tableName="m_loan_account_locks" columnName="lock_placed_on"/>
+        <dropColumn tableName="m_loan_account_locks" columnName="stacktrace"/>
+        <addColumn tableName="m_loan_account_locks">
+            <column name="stacktrace" type="TEXT"/>
+        </addColumn>
+    </changeSet>
+
+    <changeSet author="fineract" id="10" context="mysql">
+        <addColumn tableName="m_loan_account_locks">
+            <column name="lock_placed_on" type="DATETIME">
+                <constraints nullable="false"/>
+            </column>
+        </addColumn>
+    </changeSet>
+    <changeSet author="fineract" id="10" context="postgresql">
+        <addColumn tableName="m_loan_account_locks">
+            <column name="lock_placed_on" type="TIMESTAMP WITH TIME ZONE">
+                <constraints nullable="false"/>
+            </column>
+        </addColumn>
+    </changeSet>
+</databaseChangeLog>

fineract-provider/src/test/java/org/apache/fineract/cob/COBBusinessStepServiceStepDefinitions.java

@@ -0,0 +1,167 @@
+/**
+ * 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;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.base.Splitter;
+import io.cucumber.java8.En;
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TreeMap;
+import org.apache.fineract.cob.domain.BatchBusinessStep;
+import org.apache.fineract.cob.domain.BatchBusinessStepRepository;
+import org.apache.fineract.cob.exceptions.BusinessStepException;
+import org.apache.fineract.cob.loan.LoanCOBBusinessStep;
+import org.apache.fineract.infrastructure.core.domain.AbstractAuditableCustom;
+import org.apache.fineract.infrastructure.core.domain.ActionContext;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.mix.data.MixTaxonomyData;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.ListableBeanFactory;
+import org.springframework.context.ApplicationContext;
+
+public class COBBusinessStepServiceStepDefinitions implements En {
+
+    private ApplicationContext applicationContext = mock(ApplicationContext.class);
+    private ListableBeanFactory beanFactory = mock(ListableBeanFactory.class);
+    private BatchBusinessStepRepository batchBusinessStepRepository = mock(BatchBusinessStepRepository.class);
+    private final COBBusinessStepService businessStepService = new COBBusinessStepServiceImpl(batchBusinessStepRepository,
+            applicationContext, beanFactory);
+    private COBBusinessStep cobBusinessStep = mock(COBBusinessStep.class);
+    private COBBusinessStep notRegistereCobBusinessStep = mock(COBBusinessStep.class);
+    private TreeMap<Long, String> executionMap;
+    private AbstractAuditableCustom item;
+    private AbstractAuditableCustom outputItem = mock(AbstractAuditableCustom.class);
+
+    private AbstractAuditableCustom resultItem;
+
+    private HashMap<MixTaxonomyData, BigDecimal> data = new HashMap<>();
+
+    private String result;
+    private Class clazz;
+    private String jobName;
+    private BatchBusinessStep batchBusinessStep = mock(BatchBusinessStep.class);
+    private TreeMap<Long, String> resultMap;
+
+    public COBBusinessStepServiceStepDefinitions() {
+        Given("/^The COBBusinessStepService.run method with executeMap (.*)$/", (String executionMap) -> {
+            if ("null".equals(executionMap)) {
+                this.executionMap = null;
+            } else if ("".equals(executionMap)) {
+                this.executionMap = new TreeMap<>();
+            } else {
+                List<String> splitStr = Splitter.on(',').splitToList(executionMap);
+                Long key = Long.parseLong(splitStr.get(0));
+                String value = splitStr.get(1);
+                this.executionMap = new TreeMap<>();
+                this.executionMap.put(key, value);
+            }
+
+            this.item = mock(AbstractAuditableCustom.class);
+
+            lenient().when(this.applicationContext.getBean("test")).thenReturn(cobBusinessStep);
+            lenient().when(this.applicationContext.getBean("notExist")).thenThrow(BeanCreationException.class);
+            lenient().when(this.cobBusinessStep.execute(this.item)).thenReturn(outputItem);
+
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+        Given("/^The COBBusinessStepService.getCOBBusinessStepMap method with businessStepClass (.*), and jobName (.*)$/",
+                (String className, String jobName) -> {
+                    if ("null".equals(className)) {
+                        this.clazz = null;
+                    } else if ("empty".equals(className)) {
+                        this.clazz = String.class;
+                    } else if ("LoanCOBBusinessStep".equals(className)) {
+                        this.clazz = LoanCOBBusinessStep.class;
+                    } else {
+                        this.clazz = Object.class;
+                    }
+
+                    this.jobName = jobName;
+
+                    lenient().when(this.batchBusinessStepRepository.findAllByJobName("exist"))
+                            .thenReturn(Collections.singletonList(this.batchBusinessStep));
+                    lenient().when(this.batchBusinessStepRepository.findAllByJobName("notExist")).thenReturn(Collections.emptyList());
+
+                    lenient().when(this.beanFactory.getBeanNamesForType((Class<?>) null)).thenReturn(new String[] { "notExist" });
+                    lenient().when(this.beanFactory.getBeanNamesForType(Object.class)).thenReturn(new String[] { "testNotRegistered" });
+                    lenient().when(this.beanFactory.getBeanNamesForType(String.class)).thenReturn(new String[] {});
+                    lenient().when(this.beanFactory.getBeanNamesForType(LoanCOBBusinessStep.class)).thenReturn(new String[] { "test" });
+
+                    lenient().when(this.applicationContext.getBean("test")).thenReturn(this.cobBusinessStep);
+                    lenient().when(this.applicationContext.getBean("testNotRegistered")).thenReturn(this.notRegistereCobBusinessStep);
+                    lenient().when(this.applicationContext.getBean("notExist")).thenThrow(BeanCreationException.class);
+
+                    lenient().when(this.cobBusinessStep.getEnumStyledName()).thenReturn("registered");
+                    lenient().when(this.notRegistereCobBusinessStep.getEnumStyledName()).thenReturn("notRegistered");
+                    lenient().when(this.batchBusinessStep.getStepName()).thenReturn("registered");
+                    lenient().when(this.batchBusinessStep.getStepOrder()).thenReturn(1L);
+
+                    lenient().when(this.cobBusinessStep.execute(this.item)).thenReturn(outputItem);
+
+                });
+
+        When("COBBusinessStepService.run method executed", () -> {
+            resultItem = this.businessStepService.run(this.executionMap, this.item);
+        });
+
+        When("COBBusinessStepService.getCOBBusinessStepMap method executed", () -> {
+            resultMap = this.businessStepService.getCOBBusinessStepMap(this.clazz, this.jobName);
+        });
+
+        Then("The COBBusinessStepService.run result should match", () -> {
+            assertEquals(outputItem, resultItem);
+            assertEquals(ActionContext.COB, ThreadLocalContextUtil.getActionContext());
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+        Then("throw exception COBBusinessStepService.run method", () -> {
+            assertThrows(BusinessStepException.class, () -> {
+                resultItem = this.businessStepService.run(this.executionMap, this.item);
+            });
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+        Then("The COBBusinessStepService.getCOBBusinessStepMap result exception", () -> {
+            assertThrows(BeanCreationException.class, () -> {
+                this.businessStepService.getCOBBusinessStepMap(this.clazz, this.jobName);
+            });
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+        Then("The COBBusinessStepService.getCOBBusinessStepMap result should match", () -> {
+            assertEquals(1, resultMap.size());
+            assertEquals("test", resultMap.get(1L));
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+        Then("The COBBusinessStepService.getCOBBusinessStepMap result empty", () -> {
+            assertEquals(0, resultMap.size());
+            ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
+        });
+
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/common/InitialisationTaskletStepDefinitions.java

@@ -0,0 +1,67 @@
+/**
+ * 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.common;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+
+import io.cucumber.java8.En;
+import org.apache.fineract.useradministration.domain.AppUser;
+import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.security.core.context.SecurityContextHolder;
+
+public class InitialisationTaskletStepDefinitions implements En {
+
+    private AppUserRepositoryWrapper userRepository = mock(AppUserRepositoryWrapper.class);
+
+    private InitialisationTasklet initialisationTasklet = new InitialisationTasklet(userRepository);
+
+    private AppUser appUser = mock(AppUser.class);
+    private RepeatStatus resultItem;
+
+    public InitialisationTaskletStepDefinitions() {
+        Given("/^The InitialisationTasklet.execute method with action (.*)$/", (String action) -> {
+
+            if ("error".equals(action)) {
+                lenient().when(this.userRepository.fetchSystemUser()).thenThrow(new RuntimeException("fail"));
+            } else {
+                lenient().when(this.userRepository.fetchSystemUser()).thenReturn(appUser);
+            }
+
+        });
+
+        When("InitialisationTasklet.execute method executed", () -> {
+            resultItem = this.initialisationTasklet.execute(null, null);
+        });
+
+        Then("InitialisationTasklet.execute result should match", () -> {
+            assertEquals(RepeatStatus.FINISHED, resultItem);
+            assertEquals(appUser, SecurityContextHolder.getContext().getAuthentication().getPrincipal());
+        });
+
+        Then("throw exception InitialisationTasklet.execute method", () -> {
+            assertThrows(RuntimeException.class, () -> {
+                resultItem = this.initialisationTasklet.execute(null, null);
+            });
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/listener/LoanItemListenerStepDefinitions.java

@@ -0,0 +1,123 @@
+/**
+ * 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.listener;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.cucumber.java8.En;
+import java.util.List;
+import java.util.Optional;
+import org.apache.fineract.cob.domain.LoanAccountLock;
+import org.apache.fineract.cob.domain.LoanAccountLockRepository;
+import org.apache.fineract.cob.domain.LockOwner;
+import org.apache.fineract.cob.exceptions.LoanReadException;
+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.mockito.Mockito;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
+
+public class LoanItemListenerStepDefinitions implements En {
+
+    private LoanAccountLockRepository accountLockRepository = mock(LoanAccountLockRepository.class);
+    private TransactionTemplate transactionTemplate = spy(TransactionTemplate.class);
+
+    private LoanItemListener loanItemListener = new LoanItemListener(accountLockRepository, transactionTemplate);
+
+    private Exception exception;
+
+    private LoanAccountLock loanAccountLock;
+    private Loan loan = mock(Loan.class);
+
+    public LoanItemListenerStepDefinitions() {
+        Given("/^The LoanItemListener.onReadError method (.*)$/", (String action) -> {
+            ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
+            exception = new LoanReadException(1L, new RuntimeException("fail"));
+            loanAccountLock = new LoanAccountLock(1L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            when(accountLockRepository.findByLoanIdAndLockOwner(1L, LockOwner.LOAN_COB_CHUNK_PROCESSING))
+                    .thenReturn(Optional.of(loanAccountLock));
+            transactionTemplate.setTransactionManager(mock(PlatformTransactionManager.class));
+            when(loan.getId()).thenReturn(1L);
+        });
+
+        When("LoanItemListener.onReadError method executed", () -> {
+            this.loanItemListener.onReadError(exception);
+        });
+
+        Then("LoanItemListener.onReadError result should match", () -> {
+            verify(transactionTemplate, Mockito.times(1)).setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+            verify(transactionTemplate, Mockito.times(1)).execute(any());
+            verify(accountLockRepository, Mockito.times(1)).findByLoanIdAndLockOwner(1L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            assertEquals("Loan (id: 1) reading is failed", loanAccountLock.getError());
+            assertNotNull(loanAccountLock.getStacktrace());
+        });
+
+        Given("/^The LoanItemListener.onProcessError method (.*)$/", (String action) -> {
+            ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
+            exception = new LoanReadException(1L, new RuntimeException("fail"));
+            loanAccountLock = new LoanAccountLock(2L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            when(accountLockRepository.findByLoanIdAndLockOwner(2L, LockOwner.LOAN_COB_CHUNK_PROCESSING))
+                    .thenReturn(Optional.of(loanAccountLock));
+            when(loan.getId()).thenReturn(2L);
+            transactionTemplate.setTransactionManager(mock(PlatformTransactionManager.class));
+        });
+
+        When("LoanItemListener.onProcessError method executed", () -> {
+            this.loanItemListener.onProcessError(loan, exception);
+        });
+
+        Then("LoanItemListener.onProcessError result should match", () -> {
+            verify(transactionTemplate, Mockito.times(1)).setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+            verify(transactionTemplate, Mockito.times(1)).execute(any());
+            verify(accountLockRepository, Mockito.times(1)).findByLoanIdAndLockOwner(2L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            assertEquals("Loan (id: 2) processing is failed", loanAccountLock.getError());
+            assertNotNull(loanAccountLock.getStacktrace());
+        });
+
+        Given("/^The LoanItemListener.onWriteError method (.*)$/", (String action) -> {
+            ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
+            exception = new LoanReadException(3L, new RuntimeException("fail"));
+            loanAccountLock = new LoanAccountLock(3L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            when(accountLockRepository.findByLoanIdAndLockOwner(3L, LockOwner.LOAN_COB_CHUNK_PROCESSING))
+                    .thenReturn(Optional.of(loanAccountLock));
+            when(loan.getId()).thenReturn(3L);
+            transactionTemplate.setTransactionManager(mock(PlatformTransactionManager.class));
+        });
+
+        When("LoanItemListener.onWriteError method executed", () -> {
+            this.loanItemListener.onWriteError(exception, List.of(loan));
+        });
+
+        Then("LoanItemListener.onWriteError result should match", () -> {
+            verify(transactionTemplate, Mockito.times(1)).setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+            verify(transactionTemplate, Mockito.times(1)).execute(any());
+            verify(accountLockRepository, Mockito.times(1)).findByLoanIdAndLockOwner(3L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            assertEquals("Loan (id: 3) writing is failed", loanAccountLock.getError());
+            assertNotNull(loanAccountLock.getStacktrace());
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/loan/ApplyLoanLockTaskletStepDefinitions.java

@@ -0,0 +1,92 @@
+/**
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import io.cucumber.java8.En;
+import java.util.List;
+import org.apache.fineract.cob.domain.LoanAccountLock;
+import org.apache.fineract.cob.domain.LoanAccountLockRepository;
+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.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.RepeatStatus;
+
+public class ApplyLoanLockTaskletStepDefinitions implements En {
+
+    ArgumentCaptor<LoanAccountLock> valueCaptor = ArgumentCaptor.forClass(LoanAccountLock.class);
+    private LoanAccountLockRepository accountLockRepository = mock(LoanAccountLockRepository.class);
+    private ApplyLoanLockTasklet applyLoanLockTasklet = new ApplyLoanLockTasklet(accountLockRepository);
+    private RepeatStatus resultItem;
+    private StepContribution stepContribution;
+
+    public ApplyLoanLockTaskletStepDefinitions() {
+        Given("/^The ApplyLoanLockTasklet.execute method with action (.*)$/", (String action) -> {
+            ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
+            StepExecution stepExecution = new StepExecution("test", null);
+            ExecutionContext executionContext = new ExecutionContext();
+            executionContext.put(LoanCOBConstant.LOAN_IDS, List.of(1L, 2L, 3L, 4L));
+            stepExecution.setExecutionContext(executionContext);
+            this.stepContribution = new StepContribution(stepExecution);
+
+            if ("error".equals(action)) {
+                lenient().when(this.accountLockRepository.findAllByLoanIdIn(Mockito.anyList())).thenThrow(new RuntimeException("fail"));
+            } else {
+                LoanAccountLock lock1 = new LoanAccountLock(1L, LockOwner.LOAN_COB_CHUNK_PROCESSING);
+                LoanAccountLock lock2 = new LoanAccountLock(2L, LockOwner.LOAN_COB_PARTITIONING);
+                LoanAccountLock lock3 = new LoanAccountLock(3L, LockOwner.LOAN_INLINE_COB_PROCESSING);
+                List<LoanAccountLock> accountLocks = List.of(lock1, lock2, lock3);
+                lenient().when(this.accountLockRepository.findAllByLoanIdIn(Mockito.anyList())).thenReturn(accountLocks);
+            }
+
+        });
+
+        When("ApplyLoanLockTasklet.execute method executed", () -> {
+            resultItem = this.applyLoanLockTasklet.execute(stepContribution, null);
+        });
+
+        Then("ApplyLoanLockTasklet.execute result should match", () -> {
+            assertEquals(RepeatStatus.FINISHED, resultItem);
+            assertEquals(3L, ((List) stepContribution.getStepExecution().getExecutionContext()
+                    .get(LoanCOBWorkerConfiguration.ALREADY_LOCKED_LOAN_IDS)).get(0));
+            verify(this.accountLockRepository, Mockito.times(2)).save(valueCaptor.capture());
+            List<LoanAccountLock> values = valueCaptor.getAllValues();
+            assertEquals(2L, values.get(0).getLoanId());
+            assertEquals(LockOwner.LOAN_COB_CHUNK_PROCESSING, values.get(0).getLockOwner());
+            assertEquals(4L, values.get(1).getLoanId());
+            assertEquals(LockOwner.LOAN_COB_CHUNK_PROCESSING, values.get(1).getLockOwner());
+        });
+
+        Then("throw exception ApplyLoanLockTasklet.execute method", () -> {
+            assertThrows(RuntimeException.class, () -> {
+                resultItem = this.applyLoanLockTasklet.execute(stepContribution, null);
+            });
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanCOBPartitionerStepDefinitions.java

@@ -0,0 +1,113 @@
+/**
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import io.cucumber.java8.En;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import org.apache.fineract.cob.COBBusinessStepService;
+import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.infrastructure.springbatch.PropertyService;
+import org.mockito.Mockito;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.item.ExecutionContext;
+
+public class LoanCOBPartitionerStepDefinitions implements En {
+
+    PropertyService propertyService = mock(PropertyService.class);
+    COBBusinessStepService cobBusinessStepService = mock(COBBusinessStepService.class);
+    JobOperator jobOperator = mock(JobOperator.class);
+    JobExplorer jobExplorer = mock(JobExplorer.class);
+    RetrieveLoanIdService retrieveLoanIdService = mock(RetrieveLoanIdService.class);
+    private LoanCOBPartitioner loanCOBPartitioner = new LoanCOBPartitioner(propertyService, cobBusinessStepService, jobOperator,
+            jobExplorer, retrieveLoanIdService);
+
+    private TreeMap<Long, String> cobBusinessMap = new TreeMap<>();
+
+    private Map<String, ExecutionContext> resultItem;
+    private String action;
+
+    public LoanCOBPartitionerStepDefinitions() {
+        Given("/^The LoanCOBPartitioner.partition method with action (.*)$/", (String action) -> {
+
+            this.action = action;
+            lenient().when(propertyService.getPartitionSize(LoanCOBConstant.JOB_NAME)).thenReturn(2);
+            if ("empty steps".equals(action)) {
+                lenient().when(cobBusinessStepService.getCOBBusinessStepMap(LoanCOBBusinessStep.class, LoanCOBConstant.LOAN_COB_JOB_NAME))
+                        .thenReturn(new TreeMap<>());
+                lenient().when(jobExplorer.findRunningJobExecutions(JobName.LOAN_COB.name())).thenReturn(Set.of(new JobExecution(3L)));
+                lenient().when(jobOperator.stop(3L)).thenReturn(Boolean.TRUE);
+            } else if ("empty loanIds".equals(action)) {
+                cobBusinessMap.put(1L, "Business step");
+                lenient().when(cobBusinessStepService.getCOBBusinessStepMap(LoanCOBBusinessStep.class, LoanCOBConstant.LOAN_COB_JOB_NAME))
+                        .thenReturn(cobBusinessMap);
+                lenient().when(retrieveLoanIdService.retrieveLoanIds()).thenReturn(Collections.emptyList());
+                lenient().when(jobExplorer.findRunningJobExecutions(JobName.LOAN_COB.name())).thenThrow(new RuntimeException("fail"));
+            } else if ("good".equals(action)) {
+                cobBusinessMap.put(1L, "Business step");
+                lenient().when(cobBusinessStepService.getCOBBusinessStepMap(LoanCOBBusinessStep.class, LoanCOBConstant.LOAN_COB_JOB_NAME))
+                        .thenReturn(cobBusinessMap);
+                lenient().when(retrieveLoanIdService.retrieveLoanIds()).thenReturn(List.of(1L, 2L, 3L));
+            }
+        });
+
+        When("LoanCOBPartitioner.partition method executed", () -> {
+            resultItem = this.loanCOBPartitioner.partition(2);
+        });
+
+        Then("LoanCOBPartitioner.partition result should match", () -> {
+            if ("empty steps".equals(action)) {
+                verify(jobOperator, Mockito.times(1)).stop(3L);
+                assertTrue(resultItem.isEmpty());
+            } else if ("good".equals(action)) {
+                verify(jobOperator, Mockito.times(0)).stop(Mockito.anyLong());
+                assertEquals(2, resultItem.size());
+                assertTrue(resultItem.containsKey(LoanCOBPartitioner.PARTITION_PREFIX + "1"));
+                assertEquals(cobBusinessMap,
+                        resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "1").get(LoanCOBConstant.BUSINESS_STEP_MAP));
+                assertEquals(2, ((List) resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "1").get(LoanCOBConstant.LOAN_IDS)).size());
+                assertEquals(1L, ((List) resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "1").get(LoanCOBConstant.LOAN_IDS)).get(0));
+                assertEquals(2L, ((List) resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "1").get(LoanCOBConstant.LOAN_IDS)).get(1));
+                assertTrue(resultItem.containsKey(LoanCOBPartitioner.PARTITION_PREFIX + "2"));
+                assertEquals(cobBusinessMap,
+                        resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "2").get(LoanCOBConstant.BUSINESS_STEP_MAP));
+                assertEquals(1, ((List) resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "2").get(LoanCOBConstant.LOAN_IDS)).size());
+                assertEquals(3L, ((List) resultItem.get(LoanCOBPartitioner.PARTITION_PREFIX + "2").get(LoanCOBConstant.LOAN_IDS)).get(0));
+            }
+        });
+
+        Then("throw exception LoanCOBPartitioner.partition method", () -> {
+            assertThrows(RuntimeException.class, () -> {
+                resultItem = this.loanCOBPartitioner.partition(2);
+            });
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemProcessorStepDefinitions.java

@@ -0,0 +1,82 @@
+/**
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+
+import io.cucumber.java8.En;
+import java.util.TreeMap;
+import org.apache.fineract.cob.COBBusinessStepService;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+
+public class LoanItemProcessorStepDefinitions implements En {
+
+    private COBBusinessStepService cobBusinessStepService = mock(COBBusinessStepService.class);
+
+    private LoanItemProcessor loanItemProcessor = new LoanItemProcessor(cobBusinessStepService);
+
+    private Loan loan = mock(Loan.class);
+
+    private Loan loanItem;
+    private Loan processedLoan = mock(Loan.class);
+
+    private Loan resultItem;
+
+    private TreeMap<Long, String> treeMap = mock(TreeMap.class);
+
+    public LoanItemProcessorStepDefinitions() {
+        Given("/^The LoanItemProcessor.process method with item (.*)$/", (String loanItem) -> {
+
+            StepExecution stepExecution = new StepExecution("test", null);
+            ExecutionContext stepExecutionContext = new ExecutionContext();
+            stepExecutionContext.put(LoanCOBConstant.BUSINESS_STEP_MAP, treeMap);
+            stepExecution.setExecutionContext(stepExecutionContext);
+            loanItemProcessor.beforeStep(stepExecution);
+
+            if (loanItem.isEmpty()) {
+                this.loanItem = null;
+            } else {
+                this.loanItem = loan;
+            }
+
+            lenient().when(this.cobBusinessStepService.run(treeMap, null)).thenThrow(new RuntimeException("fail"));
+            lenient().when(this.cobBusinessStepService.run(treeMap, loan)).thenReturn(processedLoan);
+
+        });
+
+        When("LoanItemProcessor.process method executed", () -> {
+            resultItem = this.loanItemProcessor.process(loanItem);
+        });
+
+        Then("LoanItemProcessor.process result should match", () -> {
+            assertEquals(processedLoan, resultItem);
+        });
+
+        Then("throw exception LoanItemProcessor.process method", () -> {
+            assertThrows(RuntimeException.class, () -> {
+                resultItem = this.loanItemProcessor.process(loanItem);
+            });
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemReaderStepDefinitions.java

@@ -0,0 +1,99 @@
+/**
+ * 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.junit.Assert.assertNull;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.base.Splitter;
+import io.cucumber.java8.En;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.apache.fineract.cob.exceptions.LoanReadException;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ExecutionContext;
+
+public class LoanItemReaderStepDefinitions implements En {
+
+    private LoanRepository loanRepository = mock(LoanRepository.class);
+
+    private LoanItemReader loanItemReader = new LoanItemReader(loanRepository);
+
+    private Loan loan = mock(Loan.class);
+
+    private Loan resultItem;
+
+    public LoanItemReaderStepDefinitions() {
+        Given("/^The LoanItemReader.read method with loanIds (.*), lockedAccounts (.*)$/", (String loanIds, String lockedAccounts) -> {
+            JobExecution jobExecution = new JobExecution(1L);
+            ExecutionContext jobExecutionContext = new ExecutionContext();
+            List<Long> splitLockedAccounts;
+            if (lockedAccounts.isEmpty()) {
+                splitLockedAccounts = new ArrayList<>();
+            } else {
+                List<String> splitLockedAccountsStr = Splitter.on(',').splitToList(lockedAccounts);
+                splitLockedAccounts = splitLockedAccountsStr.stream().map(Long::parseLong).toList();
+            }
+            jobExecutionContext.put(LoanCOBWorkerConfiguration.ALREADY_LOCKED_LOAN_IDS, new ArrayList<>(splitLockedAccounts));
+            jobExecution.setExecutionContext(jobExecutionContext);
+            StepExecution stepExecution = new StepExecution("test", jobExecution);
+            ExecutionContext stepExecutionContext = new ExecutionContext();
+            List<Long> splitAccounts;
+            if (loanIds.isEmpty()) {
+                splitAccounts = new ArrayList<>();
+            } else {
+                List<String> splitStr = Splitter.on(',').splitToList(loanIds);
+                splitAccounts = splitStr.stream().map(Long::parseLong).toList();
+            }
+            stepExecutionContext.put(LoanCOBConstant.LOAN_IDS, new ArrayList<>(splitAccounts));
+            stepExecution.setExecutionContext(stepExecutionContext);
+            loanItemReader.beforeStep(stepExecution);
+
+            lenient().when(this.loanRepository.findById(0L)).thenReturn(Optional.empty());
+            lenient().when(this.loanRepository.findById(1L)).thenReturn(Optional.of(loan));
+            lenient().when(this.loanRepository.findById(-1L)).thenThrow(new RuntimeException("fail"));
+
+        });
+
+        When("LoanItemReader.read method executed", () -> {
+            resultItem = this.loanItemReader.read();
+        });
+
+        Then("The LoanItemReader.read result should match", () -> {
+            assertEquals(loan, resultItem);
+        });
+
+        Then("The LoanItemReader.read result null", () -> {
+            assertNull(resultItem);
+        });
+
+        Then("throw exception LoanItemReader.read method", () -> {
+            assertThrows(LoanReadException.class, () -> {
+                resultItem = this.loanItemReader.read();
+            });
+        });
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/cob/loan/LoanItemWriterStepDefinitions.java

@@ -0,0 +1,74 @@
+/**
+ * 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.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import io.cucumber.java8.En;
+import java.util.Collections;
+import java.util.List;
+import org.apache.fineract.cob.domain.LoanAccountLockRepository;
+import org.apache.fineract.cob.domain.LockOwner;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.mockito.Mockito;
+
+public class LoanItemWriterStepDefinitions implements En {
+
+    private LoanAccountLockRepository accountLockRepository = mock(LoanAccountLockRepository.class);
+    private LoanRepository loanRepository = mock(LoanRepository.class);
+
+    private LoanItemWriter loanItemWriter = new LoanItemWriter(accountLockRepository);
+
+    private List<Loan> items;
+
+    public LoanItemWriterStepDefinitions() {
+        Given("/^The LoanItemWriter.write method with action (.*)$/", (String action) -> {
+
+            Loan loan = mock(Loan.class);
+            lenient().when(loan.getId()).thenReturn(1L);
+            if (action.equals("error")) {
+                this.items = Collections.emptyList();
+                lenient().doThrow(new RuntimeException("fail")).when(this.accountLockRepository)
+                        .deleteByLoanIdInAndLockOwner(Collections.emptyList(), LockOwner.LOAN_COB_CHUNK_PROCESSING);
+            } else {
+                this.items = Collections.singletonList(loan);
+                lenient().doNothing().when(this.accountLockRepository).deleteByLoanIdInAndLockOwner(Mockito.anyList(), Mockito.any());
+            }
+            this.loanItemWriter.setRepository(loanRepository);
+        });
+
+        When("LoanItemWriter.write method executed", () -> {
+            this.loanItemWriter.write(items);
+        });
+
+        Then("LoanItemWriter.write result should match", () -> {
+            verify(this.accountLockRepository, Mockito.times(1)).deleteByLoanIdInAndLockOwner(Mockito.any(), Mockito.any());
+        });
+
+        Then("throw exception LoanItemWriter.write method", () -> {
+            assertThrows(RuntimeException.class, () -> {
+                this.loanItemWriter.write(items);
+            });
+        });
+    }
+}

fineract-provider/src/test/resources/features/cob/cob.feature

@@ -0,0 +1,72 @@
+#
+# 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.
+#
+
+Feature: COB
+
+  @cob
+  Scenario Outline: COB Business Step Service - run test
+    Given The COBBusinessStepService.run method with executeMap <executionMap>
+    When COBBusinessStepService.run method executed
+    Then The COBBusinessStepService.run result should match
+
+    Examples:
+      |executionMap|
+      |1,test|
+
+  @cob
+  Scenario Outline: COB Business Step Service - run test failure
+    Given The COBBusinessStepService.run method with executeMap <executionMap>
+    Then throw exception COBBusinessStepService.run method
+
+    Examples:
+      |executionMap|
+      |null|
+      |2,notExist|
+      |3,|
+
+  @cob
+  Scenario Outline: COB Business Step Service - getCOBBusinessStepMap test found bean
+    Given The COBBusinessStepService.getCOBBusinessStepMap method with businessStepClass <className>, and jobName <jobName>
+    When COBBusinessStepService.getCOBBusinessStepMap method executed
+    Then The COBBusinessStepService.getCOBBusinessStepMap result should match
+
+    Examples:
+      |className|jobName|
+      |LoanCOBBusinessStep|exist|
+
+  @cob
+  Scenario Outline: COB Business Step Service - getCOBBusinessStepMap empty
+    Given The COBBusinessStepService.getCOBBusinessStepMap method with businessStepClass <className>, and jobName <jobName>
+    When COBBusinessStepService.getCOBBusinessStepMap method executed
+    Then The COBBusinessStepService.getCOBBusinessStepMap result empty
+
+    Examples:
+      |className|jobName|
+      |LoanCOBBusinessStep|notExist|
+      |empty|exist|
+      |random|exist|
+
+  @cob
+  Scenario Outline: COB Business Step Service - getCOBBusinessStepMap no bean
+    Given The COBBusinessStepService.getCOBBusinessStepMap method with businessStepClass <className>, and jobName <jobName>
+    Then The COBBusinessStepService.getCOBBusinessStepMap result exception
+
+    Examples:
+      |className|jobName|
+      |null|notExist|

fineract-provider/src/test/resources/features/cob/common/cob.initialisation.step.feature

@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+Feature: COB Initialisation Step
+
+  @cob
+  Scenario Outline: InitialisationTasklet - run test
+    Given The InitialisationTasklet.execute method with action <action>
+    When InitialisationTasklet.execute method executed
+    Then InitialisationTasklet.execute result should match
+
+    Examples:
+      |action|
+      |good|
+
+  @cob
+  Scenario Outline: InitialisationTasklet - run test: exception
+    Given The InitialisationTasklet.execute method with action <action>
+    Then throw exception InitialisationTasklet.execute method
+
+    Examples:
+      |action|
+      |error|

fineract-provider/src/test/resources/features/cob/listener/cob.loan.item.listener.feature

@@ -0,0 +1,50 @@
+#
+# 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.
+#
+
+Feature: COB Loan item Listener
+
+  @cob
+  Scenario Outline: LoanItemListener.onReadError - run test
+    Given The LoanItemListener.onReadError method <action>
+    When LoanItemListener.onReadError method executed
+    Then LoanItemListener.onReadError result should match
+
+    Examples:
+      |action|
+      |run   |
+
+  @cob
+  Scenario Outline: LoanItemListener.onProcessError - run test
+    Given The LoanItemListener.onProcessError method <action>
+    When LoanItemListener.onProcessError method executed
+    Then LoanItemListener.onProcessError result should match
+
+    Examples:
+      |action|
+      |run   |
+
+  @cob
+  Scenario Outline: LoanItemListener.onWriteError - run test
+    Given The LoanItemListener.onWriteError method <action>
+    When LoanItemListener.onWriteError method executed
+    Then LoanItemListener.onWriteError result should match
+
+    Examples:
+      |action|
+      |run   |

fineract-provider/src/test/resources/features/cob/loan/cob.applylock.step.feature

@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+Feature: COB Apply Loan Lock Step
+
+  @cob
+  Scenario Outline: ApplyLoanLockTasklet - run test
+    Given The ApplyLoanLockTasklet.execute method with action <action>
+    When ApplyLoanLockTasklet.execute method executed
+    Then ApplyLoanLockTasklet.execute result should match
+
+    Examples:
+      |action|
+      |good|
+
+  @cob
+  Scenario Outline: ApplyLoanLockTasklet - run test: exception
+    Given The ApplyLoanLockTasklet.execute method with action <action>
+    Then throw exception ApplyLoanLockTasklet.execute method
+
+    Examples:
+      |action|
+      |error|

fineract-provider/src/test/resources/features/cob/loan/cob.loan.processor.feature

@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+Feature: COB Process
+
+  @cob
+  Scenario Outline: LoanItemProcessor - run test
+    Given The LoanItemProcessor.process method with item <item>
+    When LoanItemProcessor.process method executed
+    Then LoanItemProcessor.process result should match
+
+    Examples:
+      |item|
+      |1   |
+
+  @cob
+  Scenario Outline: LoanItemProcessor - run test: exception
+    Given The LoanItemProcessor.process method with item <item>
+    Then throw exception LoanItemProcessor.process method
+
+    Examples:
+      |item|
+      ||

fineract-provider/src/test/resources/features/cob/loan/cob.loan.reader.feature

@@ -0,0 +1,52 @@
+#
+# 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.
+#
+
+Feature: COB Reader
+
+  @cob
+  Scenario Outline: LoanItemReader - run test
+    Given The LoanItemReader.read method with loanIds <loanIds>, lockedAccounts <lockedAccounts>
+    When LoanItemReader.read method executed
+    Then The LoanItemReader.read result should match
+
+    Examples:
+      |loanIds|lockedAccounts|
+      |1      |              |
+      |1,2    |3,4           |
+
+  @cob
+  Scenario Outline: LoanItemReader - run test: null
+    Given The LoanItemReader.read method with loanIds <loanIds>, lockedAccounts <lockedAccounts>
+    When LoanItemReader.read method executed
+    Then The LoanItemReader.read result null
+
+    Examples:
+      |loanIds|lockedAccounts|
+      |       |              |
+
+  @cob
+  Scenario Outline: LoanItemReader - run test: exception
+    Given The LoanItemReader.read method with loanIds <loanIds>, lockedAccounts <lockedAccounts>
+    Then throw exception LoanItemReader.read method
+
+    Examples:
+      |loanIds|lockedAccounts|
+      |-1     |              |
+      |1      | 1            |
+      |0      |              |

fineract-provider/src/test/resources/features/cob/loan/cob.loan.writer.feature

@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+Feature: COB Writer
+
+  @cob
+  Scenario Outline: LoanItemWriter - run test
+    Given The LoanItemWriter.write method with action <action>
+    When LoanItemWriter.write method executed
+    Then LoanItemWriter.write result should match
+
+    Examples:
+      |action|
+      |good|
+
+  @cob
+  Scenario Outline: LoanItemWriter - run test: exception
+    Given The LoanItemWriter.write method with action <action>
+    Then throw exception LoanItemWriter.write method
+
+    Examples:
+      |action|
+      |error|

fineract-provider/src/test/resources/features/cob/loan/cob.partitioner.feature

@@ -0,0 +1,40 @@
+#
+# 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.
+#
+
+Feature: COB Loan partitioner Step
+
+  @cob
+  Scenario Outline: LoanCOBPartitioner - run test
+    Given The LoanCOBPartitioner.partition method with action <action>
+    When LoanCOBPartitioner.partition method executed
+    Then LoanCOBPartitioner.partition result should match
+
+    Examples:
+      |action|
+      |empty steps|
+      |good|
+
+  @cob
+  Scenario Outline: LoanCOBPartitioner - run test: exception
+    Given The LoanCOBPartitioner.partition method with action <action>
+    Then throw exception LoanCOBPartitioner.partition method
+
+    Examples:
+      |action|
+      |empty loanIds|