[FINERACT-1678] Loan COB Catch Up

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

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.cob.api;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.responses.ApiResponses;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.cob.data.IsCatchUpRunningDTO;
+import org.apache.fineract.cob.data.OldestCOBProcessedLoanDTO;
+import org.apache.fineract.cob.service.LoanCOBCatchUpService;
+import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
+import org.springframework.stereotype.Component;
+
+@Path("/loans")
+@Component
+@Tag(name = "Loan COB Catch Up", description = "")
+@RequiredArgsConstructor
+public class LoanCOBCatchUpApiResource {
+
+    private final DefaultToApiJsonSerializer<OldestCOBProcessedLoanDTO> oldestCOBProcessedLoanSerializeService;
+    private final DefaultToApiJsonSerializer<IsCatchUpRunningDTO> isCatchUpRunningSerializer;
+    private final LoanCOBCatchUpService loanCOBCatchUpService;
+
+    @GET
+    @Path("oldest-cob-closed")
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Operation(summary = "Retrieves the oldest COB processed loan", description = "Retrieves the COB business date and the oldest COB processed loan")
+    @ApiResponses({
+            @ApiResponse(responseCode = "200", description = "OK", content = @Content(schema = @Schema(implementation = LoanCOBCatchUpApiResourceSwagger.GetOldestCOBProcessedLoanResponse.class))) })
+    public String getOldestCOBProcessedLoan() {
+        OldestCOBProcessedLoanDTO response = loanCOBCatchUpService.getOldestCOBProcessedLoan();
+
+        return oldestCOBProcessedLoanSerializeService.serialize(response);
+    }
+
+    @POST
+    @Path("catch-up")
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Operation(summary = "Executes Loan COB Catch Up", description = "Executes the Loan COB job on every day from the oldest Loan to the current COB business date")
+    @ApiResponses({ @ApiResponse(responseCode = "200", description = "All loans are up to date"),
+            @ApiResponse(responseCode = "202", description = "Catch Up has been started"),
+            @ApiResponse(responseCode = "400", description = "Catch Up is already running") })
+    public Response executeLoanCOBCatchUp() {
+        if (loanCOBCatchUpService.isCatchUpRunning().isCatchUpRunning()) {
+            return Response.status(Response.Status.BAD_REQUEST).build();
+        }
+        OldestCOBProcessedLoanDTO oldestCOBProcessedLoan = loanCOBCatchUpService.getOldestCOBProcessedLoan();
+        if (oldestCOBProcessedLoan.getCobProcessedDate().equals(oldestCOBProcessedLoan.getCobBusinessDate())) {
+            return Response.status(Response.Status.OK).build();
+        }
+        loanCOBCatchUpService.executeLoanCOBCatchUp();
+
+        return Response.status(Response.Status.ACCEPTED).build();
+    }
+
+    @GET
+    @Path("is-catch-up-running")
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    @Operation(summary = "Retrieves whether Loan COB catch up is running", description = "Retrieves whether Loan COB catch up is running, and the current execution date if it is running.")
+    @ApiResponses({
+            @ApiResponse(responseCode = "200", description = "OK", content = @Content(schema = @Schema(implementation = LoanCOBCatchUpApiResourceSwagger.IsCatchUpRunningResponse.class))) })
+    public String isCatchUpRunning() {
+        IsCatchUpRunningDTO response = loanCOBCatchUpService.isCatchUpRunning();
+
+        return isCatchUpRunningSerializer.serialize(response);
+    }
+}

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

@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.cob.api;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import java.time.LocalDate;
+import java.util.List;
+
+final class LoanCOBCatchUpApiResourceSwagger {
+
+    private LoanCOBCatchUpApiResourceSwagger() {
+
+    }
+
+    @Schema(description = "GetOldestCOBProcessedLoanResponse")
+    public static final class GetOldestCOBProcessedLoanResponse {
+
+        private GetOldestCOBProcessedLoanResponse() {}
+
+        public List<Long> loanIds;
+        @Schema(example = "[2022, 9, 18]")
+        public LocalDate cobProcessedDate;
+        @Schema(example = "[2022, 9, 22]")
+        public LocalDate cobBusinessDate;
+    }
+
+    @Schema(description = "IsCatchUpRunningResponse")
+    public static final class IsCatchUpRunningResponse {
+
+        private IsCatchUpRunningResponse() {}
+
+        @Schema(example = "true")
+        public boolean isCatchUpRunning;
+        @Schema(example = "[2022, 9, 22]")
+        public LocalDate processingDate;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/data/IsCatchUpRunningDTO.java

@@ -0,0 +1,31 @@
+/**
+ * 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.data;
+
+import java.time.LocalDate;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class IsCatchUpRunningDTO {
+
+    private boolean isCatchUpRunning;
+    private LocalDate processingDate;
+}

fineract-provider/src/main/java/org/apache/fineract/cob/data/LoanIdAndLastClosedBusinessDate.java

@@ -0,0 +1,28 @@
+/**
+ * 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.data;
+
+import java.time.LocalDate;
+
+public interface LoanIdAndLastClosedBusinessDate {
+
+    Long getId();
+
+    LocalDate getLastClosedBusinessDate();
+}

fineract-provider/src/main/java/org/apache/fineract/cob/data/OldestCOBProcessedLoanDTO.java

@@ -0,0 +1,31 @@
+/**
+ * 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.data;
+
+import java.time.LocalDate;
+import java.util.List;
+import lombok.Data;
+
+@Data
+public class OldestCOBProcessedLoanDTO {
+
+    private List<Long> loanIds;
+    private LocalDate cobProcessedDate;
+    private LocalDate cobBusinessDate;
+}

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

@@ -21,6 +21,7 @@ package org.apache.fineract.cob.loan;
 public final class LoanCOBConstant {
 
     public static final String JOB_NAME = "LOAN_COB";
+    public static final String JOB_HUMAN_READABLE_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";
@@ -29,6 +30,7 @@ public final class LoanCOBConstant {
     public static final String ALREADY_LOCKED_BY_INLINE_COB_OR_PROCESSED_LOAN_IDS = "alreadyLockedOrProcessedLoanIds";
     public static final String INLINE_LOAN_COB_JOB_NAME = "INLINE_LOAN_COB";
     public static final String BUSINESS_DATE_PARAMETER_NAME = "BusinessDate";
+    public static final String IS_CATCH_UP_PARAMETER_NAME = "IS_CATCH_UP";
 
     public static final String LOAN_COB_PARTITIONER_STEP = "Loan COB partition - Step";
 

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

@@ -30,6 +30,6 @@ public class RetrieveAllNonClosedLoanIdServiceImpl implements RetrieveLoanIdServ
 
     @Override
     public List<Long> retrieveLoanIdsNDaysBehind(Long numberOfDays, LocalDate businessDate) {
-        return loanRepository.findAllNonClosedLoanIdsOneDayBehind(businessDate.minusDays(numberOfDays));
+        return loanRepository.findAllNonClosedLoanIdsByLastClosedBusinessDate(businessDate.minusDays(numberOfDays));
     }
 }

fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorService.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.service;
+
+import org.apache.fineract.infrastructure.core.domain.FineractContext;
+
+public interface AsyncLoanCOBExecutorService {
+
+    void executeLoanCOBCatchUpAsync(FineractContext context);
+}

fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java

@@ -0,0 +1,107 @@
+/**
+ * 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.service;
+
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
+import org.apache.fineract.cob.loan.LoanCOBConstant;
+import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
+import org.apache.fineract.infrastructure.core.domain.FineractContext;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO;
+import org.apache.fineract.infrastructure.jobs.domain.JobParameter;
+import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository;
+import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
+import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetailRepository;
+import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException;
+import org.apache.fineract.infrastructure.jobs.service.JobStarter;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.configuration.JobLocator;
+import org.springframework.batch.core.launch.NoSuchJobException;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorService {
+
+    private final LoanRepository loanRepository;
+    private final JobLocator jobLocator;
+    private final ScheduledJobDetailRepository scheduledJobDetailRepository;
+    private final JobStarter jobStarter;
+    private final JobParameterRepository jobParameterRepository;
+
+    @Override
+    @Async("loanCOBCatchUpThreadPoolTaskExecutor")
+    public void executeLoanCOBCatchUpAsync(FineractContext context) {
+        try {
+            ThreadLocalContextUtil.init(context);
+            LocalDate cobBusinessDate = ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE);
+            List<LoanIdAndLastClosedBusinessDate> loanIdAndLastClosedBusinessDate = loanRepository
+                    .findOldestCOBProcessedLoan(cobBusinessDate);
+            LocalDate oldestCOBProcessedDate = !loanIdAndLastClosedBusinessDate.isEmpty()
+                    ? loanIdAndLastClosedBusinessDate.get(0).getLastClosedBusinessDate()
+                    : cobBusinessDate;
+            if (oldestCOBProcessedDate.isBefore(cobBusinessDate)) {
+                executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate);
+            }
+        } catch (NoSuchJobException e) {
+            throw new JobNotFoundException(LoanCOBConstant.JOB_NAME, e);
+        } catch (JobInstanceAlreadyCompleteException | JobRestartException | JobParametersInvalidException
+                | JobExecutionAlreadyRunningException e) {
+            throw new RuntimeException(e);
+        } finally {
+            ThreadLocalContextUtil.reset();
+        }
+    }
+
+    private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate)
+            throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException,
+            JobParametersInvalidException, JobRestartException {
+        Job job = jobLocator.getJob(LoanCOBConstant.JOB_NAME);
+        ScheduledJobDetail scheduledJobDetail = scheduledJobDetailRepository.findByJobName(LoanCOBConstant.JOB_HUMAN_READABLE_NAME);
+        LocalDate executingBusinessDate = oldestCOBProcessedDate.plusDays(1);
+        while (!executingBusinessDate.isAfter(cobBusinessDate)) {
+            JobParameterDTO jobParameterDTO = new JobParameterDTO(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME,
+                    executingBusinessDate.format(DateTimeFormatter.ISO_DATE));
+            Set<JobParameterDTO> jobParameters = Collections.singleton(jobParameterDTO);
+            saveCatchUpJobParameter(scheduledJobDetail);
+            jobStarter.run(job, scheduledJobDetail, ThreadLocalContextUtil.getContext(), jobParameters);
+            executingBusinessDate = executingBusinessDate.plusDays(1);
+        }
+    }
+
+    private void saveCatchUpJobParameter(ScheduledJobDetail scheduledJobDetail) {
+        JobParameter jobParameter = new JobParameter();
+        jobParameter.setJobId(scheduledJobDetail.getId());
+        jobParameter.setParameterName(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME);
+        jobParameter.setParameterValue("1");
+        jobParameterRepository.save(jobParameter);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/service/InlineLoanCOBExecutorServiceImpl.java

@@ -24,6 +24,7 @@ import com.google.gson.Gson;
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -31,6 +32,7 @@ import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
 import org.apache.fineract.cob.domain.LoanAccountLock;
 import org.apache.fineract.cob.domain.LoanAccountLockRepository;
 import org.apache.fineract.cob.domain.LockOwner;
@@ -49,6 +51,7 @@ import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException;
 import org.apache.fineract.infrastructure.jobs.service.InlineExecutorService;
 import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
 import org.apache.fineract.infrastructure.springbatch.SpringBatchJobConstants;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Job;
@@ -73,8 +76,6 @@ import org.springframework.transaction.support.TransactionTemplate;
 public class InlineLoanCOBExecutorServiceImpl implements InlineExecutorService<Long> {
 
     private static final String JOB_EXECUTION_FAILED_MESSAGE = "Job execution failed for job with name: ";
-
-    private final GoogleGsonSerializerHelper gsonFactory;
     private final LoanAccountLockRepository loanAccountLockRepository;
     private final InlineLoanCOBExecutionDataParser dataParser;
     private final JobLauncher jobLauncher;
@@ -83,6 +84,7 @@ public class InlineLoanCOBExecutorServiceImpl implements InlineExecutorService<L
     private final TransactionTemplate transactionTemplate;
     private final CustomJobParameterRepository customJobParameterRepository;
     private final PlatformSecurityContext context;
+    private final LoanRepository loanRepository;
 
     private final Gson gson = GoogleGsonSerializerHelper.createSimpleGson();
 
@@ -96,6 +98,32 @@ public class InlineLoanCOBExecutorServiceImpl implements InlineExecutorService<L
 
     @Override
     public void execute(List<Long> loanIds, String jobName) {
+        LocalDate cobBusinessDate = ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE);
+        List<LoanIdAndLastClosedBusinessDate> loansToBeProcessed = getLoansToBeProcessed(loanIds, cobBusinessDate);
+        LocalDate executingBusinessDate = getOldestCOBBusinessDate(loansToBeProcessed).plusDays(1);
+        if (!loansToBeProcessed.isEmpty()) {
+            while (!executingBusinessDate.isAfter(cobBusinessDate)) {
+                execute(getLoanIdsToBeProcessed(loansToBeProcessed, executingBusinessDate), jobName, executingBusinessDate);
+                executingBusinessDate = executingBusinessDate.plusDays(1);
+            }
+        }
+    }
+
+    private List<Long> getLoanIdsToBeProcessed(List<LoanIdAndLastClosedBusinessDate> loansToBeProcessed, LocalDate executingBusinessDate) {
+        List<Long> loanIdsToBeProcessed = new ArrayList<>();
+        loansToBeProcessed.forEach(loan -> {
+            if (loan.getLastClosedBusinessDate() != null) {
+                if (loan.getLastClosedBusinessDate().isBefore(executingBusinessDate)) {
+                    loanIdsToBeProcessed.add(loan.getId());
+                }
+            } else {
+                loanIdsToBeProcessed.add(loan.getId());
+            }
+        });
+        return loanIdsToBeProcessed;
+    }
+
+    private void execute(List<Long> loanIds, String jobName, LocalDate businessDate) {
         lockLoanAccounts(loanIds);
         Job inlineLoanCOBJob;
         try {
@@ -104,9 +132,7 @@ public class InlineLoanCOBExecutorServiceImpl implements InlineExecutorService<L
             throw new JobNotFoundException(jobName, e);
         }
         JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(inlineLoanCOBJob)
-                .addJobParameters(new JobParameters(
-                        getJobParametersMap(loanIds, ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE))))
-                .toJobParameters();
+                .addJobParameters(new JobParameters(getJobParametersMap(loanIds, businessDate))).toJobParameters();
         JobExecution jobExecution;
         try {
             jobExecution = jobLauncher.run(inlineLoanCOBJob, jobParameters);
@@ -120,6 +146,18 @@ public class InlineLoanCOBExecutorServiceImpl implements InlineExecutorService<L
         }
     }
 
+    private LocalDate getOldestCOBBusinessDate(List<LoanIdAndLastClosedBusinessDate> loans) {
+        LoanIdAndLastClosedBusinessDate oldestLoan = loans.stream().min(Comparator
+                .comparing(LoanIdAndLastClosedBusinessDate::getLastClosedBusinessDate, Comparator.nullsLast(Comparator.naturalOrder())))
+                .orElse(null);
+        return oldestLoan != null && oldestLoan.getLastClosedBusinessDate() != null ? oldestLoan.getLastClosedBusinessDate()
+                : ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE).minusDays(1);
+    }
+
+    private List<LoanIdAndLastClosedBusinessDate> getLoansToBeProcessed(List<Long> loanIds, LocalDate cobBusinessDate) {
+        return loanRepository.findAllNonClosedLoansBehindByLoanIds(cobBusinessDate, loanIds);
+    }
+
     private List<LoanAccountLock> getLoanAccountLocks(List<Long> loanIds) {
         List<LoanAccountLock> loanAccountLocks = new ArrayList<>();
         List<Long> alreadyLockedLoanIds = new ArrayList<>();

fineract-provider/src/main/java/org/apache/fineract/cob/service/LoanCOBCatchUpService.java

@@ -0,0 +1,31 @@
+/**
+ * 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.service;
+
+import org.apache.fineract.cob.data.IsCatchUpRunningDTO;
+import org.apache.fineract.cob.data.OldestCOBProcessedLoanDTO;
+
+public interface LoanCOBCatchUpService {
+
+    OldestCOBProcessedLoanDTO getOldestCOBProcessedLoan();
+
+    void executeLoanCOBCatchUp();
+
+    IsCatchUpRunningDTO isCatchUpRunning();
+}

fineract-provider/src/main/java/org/apache/fineract/cob/service/LoanCOBCatchUpServiceImpl.java

@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.cob.service;
+
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.fineract.cob.data.IsCatchUpRunningDTO;
+import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
+import org.apache.fineract.cob.data.OldestCOBProcessedLoanDTO;
+import org.apache.fineract.cob.loan.LoanCOBConstant;
+import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
+import org.apache.fineract.infrastructure.core.domain.FineractContext;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.infrastructure.jobs.domain.JobExecutionRepository;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class LoanCOBCatchUpServiceImpl implements LoanCOBCatchUpService {
+
+    private final LoanRepository loanRepository;
+    private final AsyncLoanCOBExecutorService asyncLoanCOBExecutorService;
+    private final JobExecutionRepository jobExecutionRepository;
+    private final JobExplorer jobExplorer;
+
+    @Override
+    public OldestCOBProcessedLoanDTO getOldestCOBProcessedLoan() {
+        List<LoanIdAndLastClosedBusinessDate> loanIdAndLastClosedBusinessDate = loanRepository
+                .findOldestCOBProcessedLoan(ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE));
+        OldestCOBProcessedLoanDTO oldestCOBProcessedLoanDTO = new OldestCOBProcessedLoanDTO();
+        oldestCOBProcessedLoanDTO.setLoanIds(loanIdAndLastClosedBusinessDate.stream().map(LoanIdAndLastClosedBusinessDate::getId).toList());
+        oldestCOBProcessedLoanDTO.setCobProcessedDate(
+                loanIdAndLastClosedBusinessDate.stream().map(LoanIdAndLastClosedBusinessDate::getLastClosedBusinessDate).findFirst()
+                        .orElse(ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE)));
+        oldestCOBProcessedLoanDTO.setCobBusinessDate(ThreadLocalContextUtil.getBusinessDateByType(BusinessDateType.COB_DATE));
+        return oldestCOBProcessedLoanDTO;
+    }
+
+    @Override
+    public void executeLoanCOBCatchUp() {
+        FineractContext context = ThreadLocalContextUtil.getContext();
+        asyncLoanCOBExecutorService.executeLoanCOBCatchUpAsync(context);
+    }
+
+    @Override
+    public IsCatchUpRunningDTO isCatchUpRunning() {
+        List<Long> runningCatchUpExecutionIds = jobExecutionRepository.getRunningJobsByExecutionParameter(LoanCOBConstant.JOB_NAME,
+                LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, "1");
+        if (CollectionUtils.isNotEmpty(runningCatchUpExecutionIds)) {
+            JobExecution jobExecution = jobExplorer.getJobExecution(runningCatchUpExecutionIds.get(0));
+            String executionDateString = (String) jobExecution.getExecutionContext().get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME);
+            return new IsCatchUpRunningDTO(true, LocalDate.parse(executionDateString, DateTimeFormatter.ISO_DATE));
+        } else {
+            return new IsCatchUpRunningDTO(false, null);
+        }
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/configuration/async/CustomAsyncExceptionHandler.java

@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.infrastructure.configuration.async;
+
+import java.lang.reflect.Method;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+
+@Slf4j
+public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
+
+    @Override
+    public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
+
+        log.error("Exception message - {}", throwable.getMessage());
+        log.error("Method name - {}", method.getName());
+        for (Object param : obj) {
+            log.error("Parameter value - {}", param);
+        }
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/configuration/async/SpringAsyncConfig.java

@@ -0,0 +1,43 @@
+/**
+ * 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.configuration.async;
+
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+@Configuration
+@EnableAsync
+public class SpringAsyncConfig implements AsyncConfigurer {
+
+    @Bean(name = "loanCOBCatchUpThreadPoolTaskExecutor")
+    public ThreadPoolTaskExecutor loanCOBCatchUpThreadPoolTaskExecutor() {
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+        threadPoolTaskExecutor.setMaxPoolSize(1);
+        return threadPoolTaskExecutor;
+    }
+
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+        return new CustomAsyncExceptionHandler();
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/data/JobParametersDTO.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.infrastructure.jobs.data;
+
+import java.util.Set;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@AllArgsConstructor
+@EqualsAndHashCode
+public class JobParametersDTO {
+
+    private Set<JobParameterDTO> jobParameters;
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/domain/JobExecutionRepository.java

@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.jobs.domain;
 import static org.springframework.batch.core.BatchStatus.COMPLETED;
 import static org.springframework.batch.core.BatchStatus.FAILED;
 import static org.springframework.batch.core.BatchStatus.STARTED;
+import static org.springframework.batch.core.BatchStatus.STARTING;
 
 import java.util.List;
 import java.util.Map;
@@ -84,4 +85,17 @@ public class JobExecutionRepository {
                 "UPDATE BATCH_JOB_EXECUTION SET STATUS = :status, START_TIME = null, END_TIME = null WHERE JOB_EXECUTION_ID = :jobExecutionId",
                 Map.of("status", FAILED.name(), "jobExecutionId", stuckJobId));
     }
+
+    public List<Long> getRunningJobsByExecutionParameter(String jobName, String parameterKeyName, String parameterValue) {
+        return namedParameterJdbcTemplate.queryForList("SELECT bje.JOB_EXECUTION_ID FROM BATCH_JOB_INSTANCE bji "
+                + "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
+                + "INNER JOIN BATCH_JOB_EXECUTION_PARAMS bjep ON bje.JOB_EXECUTION_ID = bjep.JOB_EXECUTION_ID "
+                + "WHERE bje.STATUS IN (:statuses) AND bji.JOB_NAME = :jobName AND bjep.KEY_NAME = :parameterKeyName AND bjep.STRING_VAL = :parameterValue AND bje.JOB_INSTANCE_ID NOT IN ("
+                + "SELECT bje.JOB_INSTANCE_ID FROM BATCH_JOB_INSTANCE bji "
+                + "INNER JOIN BATCH_JOB_EXECUTION bje ON bji.JOB_INSTANCE_ID = bje.JOB_INSTANCE_ID "
+                + "WHERE bje.STATUS = :completedStatus AND bji.JOB_NAME = :jobName)",
+                Map.of("statuses", List.of(STARTED.name(), STARTING.name()), "jobName", jobName, "completedStatus", COMPLETED.name(),
+                        "parameterKeyName", parameterKeyName, "parameterValue", parameterValue),
+                Long.class);
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/domain/ScheduledJobDetailRepository.java

@@ -45,4 +45,6 @@ public interface ScheduledJobDetailRepository
     @Query("select jobDetail from ScheduledJobDetail jobDetail where jobDetail.nodeId = :nodeId or jobDetail.nodeId = 0")
     List<ScheduledJobDetail> findAllJobs(@Param("nodeId") Integer nodeId);
 
+    ScheduledJobDetail findByJobName(String jobName);
+
 }

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

@@ -21,6 +21,7 @@ package org.apache.fineract.portfolio.loanaccount.domain;
 import java.time.LocalDate;
 import java.util.Collection;
 import java.util.List;
+import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate;
 import org.apache.fineract.infrastructure.core.domain.ExternalId;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
@@ -72,7 +73,7 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
 
     String FIND_ALL_NON_CLOSED = "select loan.id from Loan loan where loan.loanStatus in (100,200,300,303,304)";
 
-    String FIND_ALL_NON_CLOSED_ONE_DAY_BEHIND = "select loan.id from Loan loan where loan.loanStatus in (100,200,300,303,304) and (:last_closed_business_date = loan.lastClosedBusinessDate or loan.lastClosedBusinessDate is NULL)";
+    String FIND_ALL_NON_CLOSED_LOANS_BY_LAST_CLOSED_BUSINESS_DATE = "select loan.id from Loan loan where loan.loanStatus in (100,200,300,303,304) and (:businessDate = loan.lastClosedBusinessDate or loan.lastClosedBusinessDate is NULL)";
 
     String FIND_NON_CLOSED_LOAN_THAT_BELONGS_TO_CLIENT = "select loan from Loan loan where loan.id = :loanId and loan.loanStatus = 300 and loan.client.id = :clientId";
 
@@ -80,6 +81,13 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
 
     String FIND_ID_BY_EXTERNAL_ID = "SELECT loan.id FROM Loan loan WHERE loan.externalId = :externalId";
 
+    // should follow the logic of `FIND_ALL_NON_CLOSED_LOANS_BY_LAST_CLOSED_BUSINESS_DATE` query
+    String FIND_OLDEST_COB_PROCESSED_LOAN = "select loan.id, loan.lastClosedBusinessDate from Loan loan where loan.lastClosedBusinessDate = (select min(l.lastClosedBusinessDate) from Loan l where loan.loanStatus in (100,200,300,303,304) and l"
+            + ".lastClosedBusinessDate <> " + ":cobBusinessDate)";
+
+    String FIND_ALL_NON_CLOSED_LOANS_BEHIND_BY_LOAN_IDS = "select loan.id, loan.lastClosedBusinessDate from Loan loan where loan.id IN :loanIds and loan.loanStatus in (100,200,300,303,304) and (loan.lastClosedBusinessDate <> :cobBusinessDate or "
+            + "loan.lastClosedBusinessDate is null)";
+
     @Query(FIND_GROUP_LOANS_DISBURSED_AFTER)
     List<Loan> getGroupLoansDisbursedAfter(@Param("disbursementDate") LocalDate disbursementDate, @Param("groupId") Long groupId,
             @Param("loanType") Integer loanType);
@@ -176,7 +184,14 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
     @Query(FIND_ID_BY_EXTERNAL_ID)
     Long findIdByExternalId(@Param("externalId") ExternalId externalId);
 
-    @Query(FIND_ALL_NON_CLOSED_ONE_DAY_BEHIND)
-    List<Long> findAllNonClosedLoanIdsOneDayBehind(@Param("last_closed_business_date") LocalDate businessDate);
+    @Query(FIND_ALL_NON_CLOSED_LOANS_BY_LAST_CLOSED_BUSINESS_DATE)
+    List<Long> findAllNonClosedLoanIdsByLastClosedBusinessDate(@Param("businessDate") LocalDate businessDate);
+
+    @Query(FIND_ALL_NON_CLOSED_LOANS_BEHIND_BY_LOAN_IDS)
+    List<LoanIdAndLastClosedBusinessDate> findAllNonClosedLoansBehindByLoanIds(@Param("cobBusinessDate") LocalDate cobBusinessDate,
+            @Param("loanIds") List<Long> loanIds);
+
+    @Query(FIND_OLDEST_COB_PROCESSED_LOAN)
+    List<LoanIdAndLastClosedBusinessDate> findOldestCOBProcessedLoan(@Param("cobBusinessDate") LocalDate cobBusinessDate);
 
 }