FINERACT-2421: Improved Spring Batch context handling

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

@@ -28,12 +28,12 @@ import java.util.List;
 import java.util.Objects;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.fineract.cob.common.CustomJobParameterResolver;
 import org.apache.fineract.cob.converter.COBParameterConverter;
 import org.apache.fineract.cob.data.COBParameter;
 import org.apache.fineract.cob.domain.LoanAccountLock;
 import org.apache.fineract.cob.domain.LockOwner;
 import org.apache.fineract.cob.exceptions.LoanLockCannotBeAppliedException;
+import org.apache.fineract.cob.resolver.CatchUpFlagResolver;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.scope.context.ChunkContext;
@@ -53,7 +53,6 @@ public class ApplyLoanLockTasklet implements Tasklet {
     private final FineractProperties fineractProperties;
     private final LoanLockingService loanLockingService;
     private final RetrieveLoanIdService retrieveLoanIdService;
-    private final CustomJobParameterResolver customJobParameterResolver;
     private final TransactionTemplate transactionTemplate;
 
     @Override
@@ -63,6 +62,7 @@ public class ApplyLoanLockTasklet implements Tasklet {
         ExecutionContext executionContext = contribution.getStepExecution().getExecutionContext();
         long numberOfExecutions = contribution.getStepExecution().getCommitCount();
         COBParameter loanCOBParameter = COBParameterConverter.convert(executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER));
+        boolean isCatchUp = CatchUpFlagResolver.resolve(contribution.getStepExecution());
         List<Long> loanIds;
         if (Objects.isNull(loanCOBParameter)
                 || (Objects.isNull(loanCOBParameter.getMinAccountId()) && Objects.isNull(loanCOBParameter.getMaxAccountId()))
@@ -70,10 +70,7 @@ public class ApplyLoanLockTasklet implements Tasklet {
             loanIds = Collections.emptyList();
         } else {
             loanIds = new ArrayList<>(
-                    retrieveLoanIdService.retrieveAllNonClosedLoansByLastClosedBusinessDateAndMinAndMaxLoanId(loanCOBParameter,
-                            customJobParameterResolver
-                                    .getCustomJobParameterById(contribution.getStepExecution(), LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME)
-                                    .map(Boolean::parseBoolean).orElse(false)));
+                    retrieveLoanIdService.retrieveAllNonClosedLoansByLastClosedBusinessDateAndMinAndMaxLoanId(loanCOBParameter, isCatchUp));
         }
         List<List<Long>> loanIdPartitions = Lists.partition(loanIds, getInClauseParameterSizeLimit());
         List<LoanAccountLock> accountLocks = new ArrayList<>();

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

@@ -32,6 +32,7 @@ public final class LoanCOBConstant extends COBConstant {
     public static final String LOAN_IDS_PARAMETER_NAME = "LoanIds";
 
     public static final String LOAN_COB_PARTITIONER_STEP = "Loan COB partition - Step";
+    public static final String PARTITION_KEY = "partition";
 
     private LoanCOBConstant() {
 

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

@@ -20,20 +20,17 @@ package org.apache.fineract.cob.loan;
 
 import static org.apache.fineract.cob.loan.LoanCOBConstant.JOB_NAME;
 
-import java.util.List;
 import org.apache.fineract.cob.COBBusinessStepService;
 import org.apache.fineract.cob.common.CustomJobParameterResolver;
 import org.apache.fineract.cob.conditions.BatchManagerCondition;
 import org.apache.fineract.cob.listener.COBExecutionListenerRunner;
-import org.apache.fineract.cob.listener.JobExecutionContextCopyListener;
 import org.apache.fineract.infrastructure.event.business.service.BusinessEventNotifierService;
 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.JobScope;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.configuration.annotation.StepScope;
-import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.job.builder.JobBuilder;
 import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.launch.support.RunIdIncrementer;
@@ -43,6 +40,7 @@ import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
@@ -70,8 +68,6 @@ public class LoanCOBManagerConfiguration {
     @Autowired
     private JobOperator jobOperator;
     @Autowired
-    private JobExplorer jobExplorer;
-    @Autowired
     private ApplicationContext applicationContext;
     @Autowired
     private RetrieveLoanIdService retrieveLoanIdService;
@@ -82,17 +78,16 @@ public class LoanCOBManagerConfiguration {
 
     @Bean
     @StepScope
-    public LoanCOBPartitioner partitioner() {
-        return new LoanCOBPartitioner(propertyService, cobBusinessStepService, retrieveLoanIdService, jobOperator, jobExplorer,
+    public LoanCOBPartitioner partitioner(@Value("#{stepExecution}") StepExecution stepExecution) {
+        return new LoanCOBPartitioner(propertyService, cobBusinessStepService, retrieveLoanIdService, jobOperator, stepExecution,
                 LoanCOBConstant.NUMBER_OF_DAYS_BEHIND);
     }
 
-    @Bean
-    public Step loanCOBStep() {
+    @Bean("loanCOBStep")
+    public Step loanCOBStep(LoanCOBPartitioner partitioner) {
         return stepBuilderFactory.get(LoanCOBConstant.LOAN_COB_PARTITIONER_STEP)
-                .partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner()).pollInterval(propertyService.getPollInterval(JOB_NAME))
-                .listener(new JobExecutionContextCopyListener(List.of("BusinessDate", "IS_CATCH_UP"))).outputChannel(outboundRequests)
-                .build();
+                .partitioner(LoanCOBConstant.LOAN_COB_WORKER_STEP, partitioner).pollInterval(propertyService.getPollInterval(JOB_NAME))
+                .outputChannel(outboundRequests).build();
     }
 
     @Bean
@@ -108,23 +103,21 @@ public class LoanCOBManagerConfiguration {
     }
 
     @Bean
-    @JobScope
     public ResolveLoanCOBCustomJobParametersTasklet resolveCustomJobParametersTasklet() {
         return new ResolveLoanCOBCustomJobParametersTasklet(customJobParameterResolver);
     }
 
     @Bean
-    @JobScope
     public StayedLockedLoansTasklet stayedLockedTasklet() {
         return new StayedLockedLoansTasklet(businessEventNotifierService, retrieveLoanIdService);
     }
 
     @Bean(name = "loanCOBJob")
-    public Job loanCOBJob() {
+    public Job loanCOBJob(LoanCOBPartitioner partitioner) {
         return new JobBuilder(JobName.LOAN_COB.name(), jobRepository) //
                 .listener(new COBExecutionListenerRunner(applicationContext, JobName.LOAN_COB.name())) //
                 .start(resolveCustomJobParametersStep()) //
-                .next(loanCOBStep()).next(stayedLockedStep()) //
+                .next(loanCOBStep(partitioner)).next(stayedLockedStep()) //
                 .incrementer(new RunIdIncrementer()) //
                 .build();
     }

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

@@ -25,22 +25,20 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.cob.COBBusinessStepService;
 import org.apache.fineract.cob.data.BusinessStepNameAndOrder;
 import org.apache.fineract.cob.data.COBParameter;
 import org.apache.fineract.cob.data.COBPartition;
-import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.cob.resolver.BusinessDateResolver;
+import org.apache.fineract.cob.resolver.CatchUpFlagResolver;
 import org.apache.fineract.infrastructure.springbatch.PropertyService;
-import org.springframework.batch.core.JobExecution;
-import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.launch.JobExecutionNotRunningException;
 import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.launch.NoSuchJobExecutionException;
 import org.springframework.batch.core.partition.support.Partitioner;
 import org.springframework.batch.item.ExecutionContext;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.lang.NonNull;
 import org.springframework.util.StopWatch;
 
@@ -54,17 +52,9 @@ public class LoanCOBPartitioner implements Partitioner {
     private final COBBusinessStepService cobBusinessStepService;
     private final RetrieveLoanIdService retrieveLoanIdService;
     private final JobOperator jobOperator;
-    private final JobExplorer jobExplorer;
-
+    private final StepExecution stepExecution;
     private final Long numberOfDays;
 
-    @Value("#{stepExecutionContext['BusinessDate']}")
-    @Setter
-    private LocalDate businessDate;
-    @Value("#{stepExecutionContext['IS_CATCH_UP']}")
-    @Setter
-    private Boolean isCatchUp;
-
     @NonNull
     @Override
     public Map<String, ExecutionContext> partition(int gridSize) {
@@ -79,45 +69,49 @@ public class LoanCOBPartitioner implements Partitioner {
             stopJobExecution();
             return Map.of();
         }
+        LocalDate businessDate = BusinessDateResolver.resolve(stepExecution);
+        boolean isCatchUp = CatchUpFlagResolver.resolve(stepExecution);
         StopWatch sw = new StopWatch();
         sw.start();
         List<COBPartition> loanCOBPartitions = new ArrayList<>(
-                retrieveLoanIdService.retrieveLoanCOBPartitions(numberOfDays, businessDate, isCatchUp != null && isCatchUp, partitionSize));
+                retrieveLoanIdService.retrieveLoanCOBPartitions(numberOfDays, businessDate, isCatchUp, partitionSize));
         sw.stop();
         // if there is no loan to be closed, we still would like to create at least one partition
 
-        if (loanCOBPartitions.size() == 0) {
+        if (loanCOBPartitions.isEmpty()) {
             loanCOBPartitions.add(new COBPartition(0L, 0L, 1L, 0L));
         }
         log.info(
                 "LoanCOBPartitioner found {} loans to be processed as part of COB. {} partitions were created using partition size {}. RetrieveLoanCOBPartitions was executed in {} ms.",
                 getLoanCount(loanCOBPartitions), loanCOBPartitions.size(), partitionSize, sw.getTotalTimeMillis());
-        return loanCOBPartitions.stream()
-                .collect(Collectors.toMap(l -> PARTITION_PREFIX + l.getPageNo(), l -> createNewPartition(cobBusinessSteps, l)));
+        return loanCOBPartitions.stream().collect(Collectors.toMap(l -> PARTITION_PREFIX + l.getPageNo(),
+                l -> createNewPartition(cobBusinessSteps, l, businessDate, isCatchUp)));
     }
 
     private long getLoanCount(List<COBPartition> loanCOBPartitions) {
         return loanCOBPartitions.stream().map(COBPartition::getCount).reduce(0L, Long::sum);
     }
 
-    private ExecutionContext createNewPartition(Set<BusinessStepNameAndOrder> cobBusinessSteps, COBPartition loanCOBPartition) {
+    private ExecutionContext createNewPartition(Set<BusinessStepNameAndOrder> cobBusinessSteps, COBPartition loanCOBPartition,
+            LocalDate businessDate, boolean isCatchUp) {
         ExecutionContext executionContext = new ExecutionContext();
         executionContext.put(LoanCOBConstant.BUSINESS_STEPS, cobBusinessSteps);
         executionContext.put(LoanCOBConstant.LOAN_COB_PARAMETER,
                 new COBParameter(loanCOBPartition.getMinId(), loanCOBPartition.getMaxId()));
-        executionContext.put("partition", PARTITION_PREFIX + loanCOBPartition.getPageNo());
+        executionContext.put(LoanCOBConstant.PARTITION_KEY, PARTITION_PREFIX + loanCOBPartition.getPageNo());
+        executionContext.put(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME, businessDate.toString());
+        executionContext.put(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, Boolean.toString(isCatchUp));
         return executionContext;
     }
 
     private void stopJobExecution() {
-        Set<JobExecution> runningJobExecutions = jobExplorer.findRunningJobExecutions(JobName.LOAN_COB.name());
-        for (JobExecution jobExecution : runningJobExecutions) {
-            try {
-                jobOperator.stop(jobExecution.getId());
-            } catch (NoSuchJobExecutionException | JobExecutionNotRunningException e) {
-                log.error("There is no running execution for the given execution ID. Execution ID: {}", jobExecution.getId());
-                throw new RuntimeException(e);
-            }
+        Long jobId = stepExecution.getJobExecution().getId();
+        try {
+            jobOperator.stop(jobId);
+        } catch (NoSuchJobExecutionException | JobExecutionNotRunningException e) {
+            log.error("There is no running execution for the given execution ID. Execution ID: {}", jobId);
+            throw new RuntimeException(e);
         }
+
     }
 }

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

@@ -19,7 +19,6 @@
 package org.apache.fineract.cob.loan;
 
 import org.apache.fineract.cob.COBBusinessStepService;
-import org.apache.fineract.cob.common.CustomJobParameterResolver;
 import org.apache.fineract.cob.common.InitialisationTasklet;
 import org.apache.fineract.cob.common.ResetContextTasklet;
 import org.apache.fineract.cob.conditions.BatchWorkerCondition;
@@ -81,21 +80,18 @@ public class LoanCOBWorkerConfiguration {
     @Autowired
     private LoanLockingService loanLockingService;
 
-    @Autowired
-    private CustomJobParameterResolver customJobParameterResolver;
-
     @Bean(name = LoanCOBConstant.LOAN_COB_WORKER_STEP)
-    public Step loanCOBWorkerStep() {
-        return stepBuilderFactory.get("Loan COB worker - Step").inputChannel(inboundRequests).flow(flow()).build();
+    public Step loanCOBWorkerStep(Flow cobFlow) {
+        return stepBuilderFactory.get("Loan COB worker - Step").inputChannel(inboundRequests).flow(cobFlow).build();
     }
 
-    @Bean
-    public Flow flow() {
-        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep(null)).next(applyLockStep(null)).next(loanBusinessStep(null, null))
-                .next(resetContextStep(null)).build();
+    @Bean("cobFlow")
+    public Flow flow(Step initialisationStep, Step applyLockStep, Step loanBusinessStep, Step resetContextStep) {
+        return new FlowBuilder<Flow>("cobFlow").start(initialisationStep).next(applyLockStep).next(loanBusinessStep).next(resetContextStep)
+                .build();
     }
 
-    @Bean
+    @Bean("initialisationStep")
     @StepScope
     public Step initialisationStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
         return new StepBuilder("Initialisation - Step:" + partitionName, jobRepository).tasklet(initialiseContext(), transactionManager)
@@ -118,7 +114,7 @@ public class LoanCOBWorkerConfiguration {
         return taskExecutor;
     }
 
-    @Bean
+    @Bean("loanBusinessStep")
     @StepScope
     public Step loanBusinessStep(@Value("#{stepExecutionContext['partition']}") String partitionName, TaskExecutor cobTaskExecutor) {
         SimpleStepBuilder<Loan, Loan> stepBuilder = new StepBuilder("Loan Business - Step:" + partitionName, jobRepository)
@@ -141,13 +137,13 @@ public class LoanCOBWorkerConfiguration {
         return stepBuilder.build();
     }
 
-    @Bean
+    @Bean("applyLockStep")
     @StepScope
     public Step applyLockStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
         return new StepBuilder("Apply lock - Step:" + partitionName, jobRepository).tasklet(applyLock(), transactionManager).build();
     }
 
-    @Bean
+    @Bean("resetContextStep")
     @StepScope
     public Step resetContextStep(@Value("#{stepExecutionContext['partition']}") String partitionName) {
         return new StepBuilder("Reset context - Step:" + partitionName, jobRepository).tasklet(resetContext(), transactionManager).build();
@@ -165,8 +161,7 @@ public class LoanCOBWorkerConfiguration {
 
     @Bean
     public ApplyLoanLockTasklet applyLock() {
-        return new ApplyLoanLockTasklet(fineractProperties, loanLockingService, retrieveLoanIdService, customJobParameterResolver,
-                transactionTemplate);
+        return new ApplyLoanLockTasklet(fineractProperties, loanLockingService, retrieveLoanIdService, transactionTemplate);
     }
 
     @Bean
@@ -177,7 +172,7 @@ public class LoanCOBWorkerConfiguration {
     @Bean
     @StepScope
     public LoanItemReader cobWorkerItemReader() {
-        return new LoanItemReader(loanRepository, retrieveLoanIdService, customJobParameterResolver, loanLockingService);
+        return new LoanItemReader(loanRepository, retrieveLoanIdService, loanLockingService);
     }
 
     @Bean

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

@@ -24,11 +24,11 @@ import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.LinkedBlockingQueue;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.fineract.cob.common.CustomJobParameterResolver;
 import org.apache.fineract.cob.converter.COBParameterConverter;
 import org.apache.fineract.cob.data.COBParameter;
 import org.apache.fineract.cob.domain.LoanAccountLock;
 import org.apache.fineract.cob.domain.LockOwner;
+import org.apache.fineract.cob.resolver.CatchUpFlagResolver;
 import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.annotation.BeforeStep;
@@ -39,14 +39,12 @@ import org.springframework.lang.NonNull;
 public class LoanItemReader extends AbstractLoanItemReader {
 
     private final RetrieveLoanIdService retrieveLoanIdService;
-    private final CustomJobParameterResolver customJobParameterResolver;
     private final LoanLockingService loanLockingService;
 
     public LoanItemReader(LoanRepository loanRepository, RetrieveLoanIdService retrieveLoanIdService,
-            CustomJobParameterResolver customJobParameterResolver, LoanLockingService loanLockingService) {
+            LoanLockingService loanLockingService) {
         super(loanRepository);
         this.retrieveLoanIdService = retrieveLoanIdService;
-        this.customJobParameterResolver = customJobParameterResolver;
         this.loanLockingService = loanLockingService;
     }
 
@@ -56,15 +54,15 @@ public class LoanItemReader extends AbstractLoanItemReader {
         ExecutionContext executionContext = stepExecution.getExecutionContext();
         COBParameter loanCOBParameter = COBParameterConverter.convert(executionContext.get(LoanCOBConstant.LOAN_COB_PARAMETER));
         List<Long> loanIds;
+        boolean isCatchUp = CatchUpFlagResolver.resolve(stepExecution);
         if (Objects.isNull(loanCOBParameter)
                 || (Objects.isNull(loanCOBParameter.getMinAccountId()) && Objects.isNull(loanCOBParameter.getMaxAccountId()))
                 || (loanCOBParameter.getMinAccountId().equals(0L) && loanCOBParameter.getMaxAccountId().equals(0L))) {
             loanIds = Collections.emptyList();
         } else {
             loanIds = retrieveLoanIdService.retrieveAllNonClosedLoansByLastClosedBusinessDateAndMinAndMaxLoanId(loanCOBParameter,
-                    customJobParameterResolver.getCustomJobParameterById(stepExecution, LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME)
-                            .map(Boolean::parseBoolean).orElse(false));
-            if (loanIds.size() > 0) {
+                    isCatchUp);
+            if (!loanIds.isEmpty()) {
                 List<Long> lockedByCOBChunkProcessingAccountIds = getLoanIdsLockedWithChunkProcessingLock(loanIds);
                 loanIds.retainAll(lockedByCOBChunkProcessingAccountIds);
             }
@@ -73,8 +71,8 @@ public class LoanItemReader extends AbstractLoanItemReader {
     }
 
     private List<Long> getLoanIdsLockedWithChunkProcessingLock(List<Long> loanIds) {
-        List<LoanAccountLock> accountLocks = new ArrayList<>();
-        accountLocks.addAll(loanLockingService.findAllByLoanIdInAndLockOwner(loanIds, LockOwner.LOAN_COB_CHUNK_PROCESSING));
+        List<LoanAccountLock> accountLocks = new ArrayList<>(
+                loanLockingService.findAllByLoanIdInAndLockOwner(loanIds, LockOwner.LOAN_COB_CHUNK_PROCESSING));
         return accountLocks.stream().map(LoanAccountLock::getLoanId).toList();
     }
 }

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

@@ -36,23 +36,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
 @Slf4j
 public class LoanLockingServiceImpl implements LoanLockingService {
 
-    private static final String NORMAL_LOAN_INSERT = """
-                INSERT INTO m_loan_account_locks (loan_id, version, lock_owner, lock_placed_on, lock_placed_on_cob_business_date)
-                SELECT loan.id, ?, ?, ?, ? FROM m_loan loan
-                    WHERE loan.id NOT IN (SELECT loan_id FROM m_loan_account_locks)
-                    AND loan.id BETWEEN ? AND ?
-                    AND loan.loan_status_id IN (100,200,300,303,304)
-                    AND (? = loan.last_closed_business_date OR loan.last_closed_business_date IS NULL)
-            """;
-    private static final String CATCH_UP_LOAN_INSERT = """
-                INSERT INTO m_loan_account_locks (loan_id, version, lock_owner, lock_placed_on, lock_placed_on_cob_business_date)
-                SELECT loan.id, ?, ?, ?, ? FROM m_loan loan
-                    WHERE loan.id NOT IN (SELECT loan_id FROM m_loan_account_locks)
-                    AND loan.id BETWEEN ? AND ?
-                    AND loan.loan_status_id IN (100,200,300,303,304)
-                    AND (? = loan.last_closed_business_date)
-            """;
-
     private static final String BATCH_LOAN_LOCK_INSERT = """
                 INSERT INTO m_loan_account_locks (loan_id, version, lock_owner, lock_placed_on, lock_placed_on_cob_business_date) VALUES (?,?,?,?,?)
             """;

fineract-provider/src/main/java/org/apache/fineract/cob/resolver/BusinessDateResolver.java

@@ -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.
+ */
+package org.apache.fineract.cob.resolver;
+
+import java.time.LocalDate;
+import org.apache.fineract.cob.loan.LoanCOBConstant;
+import org.springframework.batch.core.StepExecution;
+
+public final class BusinessDateResolver {
+
+    private BusinessDateResolver() {}
+
+    public static LocalDate resolve(StepExecution stepExecution) {
+        Object bd = stepExecution.getJobExecution().getExecutionContext().get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME);
+        return switch (bd) {
+            case null -> throw new IllegalStateException(
+                    "Missing BusinessDate in JobExecutionContext for jobExecutionId=" + stepExecution.getJobExecution().getId());
+            case String bdStr -> LocalDate.parse(bdStr);
+            case LocalDate bdLocalDate -> bdLocalDate;
+            default -> throw new IllegalStateException("BusinessDate value is unrecognizable: " + bd);
+        };
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/resolver/CatchUpFlagResolver.java

@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.cob.resolver;
+
+import org.apache.fineract.cob.loan.LoanCOBConstant;
+import org.springframework.batch.core.StepExecution;
+
+public final class CatchUpFlagResolver {
+
+    private CatchUpFlagResolver() {}
+
+    public static boolean resolve(StepExecution stepExecution) {
+        Object isCatchUp = stepExecution.getJobExecution().getExecutionContext().get(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME);
+        return switch (isCatchUp) {
+            case null -> false;
+            case String isCatchUpStr -> Boolean.parseBoolean(isCatchUpStr);
+            case Boolean b -> b;
+            default -> throw new IllegalStateException("isCatchUp value is unrecognizable: " + isCatchUp);
+        };
+    }
+}

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

@@ -45,8 +45,7 @@ public class InputChannelInterceptor implements ExecutorChannelInterceptor {
     @Override
     public void afterMessageHandled(@NonNull final Message<?> message, @NonNull final MessageChannel channel,
             @NonNull final MessageHandler handler, final Exception ex) {
-        log.debug("Cleaning up ThreadLocal context after message handling");
-        ThreadLocalContextUtil.reset();
+        afterHandleMessage();
     }
 
     public Message<StepExecutionRequest> beforeHandleMessage(Message<?> message) {
@@ -54,9 +53,14 @@ public class InputChannelInterceptor implements ExecutorChannelInterceptor {
     }
 
     public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) {
-        log.debug("Initializing ThreadLocal context for message handling");
+        log.debug("Initializing ThreadLocal context for message handling: {}", contextualMessage);
         ThreadLocalContextUtil.init(contextualMessage.getContext());
         ThreadLocalContextUtil.setActionContext(ActionContext.COB);
         return contextualMessage.getStepExecutionRequest();
     }
+
+    public void afterHandleMessage() {
+        log.debug("Cleaning up ThreadLocal context after message handling");
+        ThreadLocalContextUtil.reset();
+    }
 }

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

@@ -58,6 +58,8 @@ public class JmsBatchWorkerMessageListener implements MessageListener, Initializ
             stepExecutionRequestHandler.handle(requestMessage.getPayload());
         } catch (Exception e) {
             log.error("Exception while processing JMS message", e);
+        } finally {
+            inputInterceptor.afterHandleMessage();
         }
 
         try {

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

@@ -50,6 +50,8 @@ public class KafkaRemoteMessageListener {
             stepExecutionRequestHandler.handle(stepExecutionRequest);
         } catch (Exception e) {
             log.error("Exception while processing Kafka message", e);
+        } finally {
+            inputInterceptor.afterHandleMessage();
         }
         acknowledgment.acknowledge();
         log.debug("Message was acknowledged {}", acknowledgment);