[FINERACT-1678] Loan COB job skeleton

fineract-provider/dependencies.gradle

@@ -32,6 +32,10 @@ dependencies {
             'org.springframework.boot:spring-boot-starter-cache',
             'org.springframework.boot:spring-boot-starter-oauth2-resource-server',
             'org.springframework.boot:spring-boot-starter-actuator',
+            'org.springframework.boot:spring-boot-starter-batch',
+            'org.springframework.batch:spring-batch-integration',
+            'org.springframework.boot:spring-boot-starter-integration',
+            'org.springframework.integration:spring-integration-event',
 
             'org.glassfish.jersey.media:jersey-media-multipart:2.36',
 
@@ -129,9 +133,6 @@ dependencies {
         exclude group: 'xml-apis'
     }
 
-    implementation ('org.springframework.boot:spring-boot-starter-batch') {
-    }
-
     runtimeOnly('org.glassfish.jaxb:jaxb-runtime') {
         exclude group: 'com.sun.activation'
     }

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

@@ -0,0 +1,36 @@
+/**
+ * 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 org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+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 {
+
+    @Override
+    public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
+        COBMessage castedMessage = COBMessage.class.cast(message.getPayload());
+        ThreadLocalContextUtil.init(castedMessage.getContext());
+        return new GenericMessage<>(castedMessage.getStepExecutionRequest());
+    }
+}

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

@@ -0,0 +1,55 @@
+/**
+ * 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 org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
+import org.springframework.integration.handler.LoggingHandler;
+
+@Configuration
+@EnableBatchIntegration
+@ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
+public class COBManagerConfig {
+
+    @Bean
+    public IntegrationFlow outboundFlow() {
+        ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
+        return IntegrationFlows.from(outboundRequests()) //
+                .intercept(outputInterceptor()) //
+                .log(LoggingHandler.Level.DEBUG) //
+                .handle(handler) //
+                .get(); //
+    }
+
+    @Bean
+    public DirectChannel outboundRequests() {
+        return new DirectChannel();
+    }
+
+    @Bean
+    public COBOutputChannelInterceptor outputInterceptor() {
+        return new COBOutputChannelInterceptor();
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/COBMessage.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;
+
+import java.io.Serializable;
+import lombok.Data;
+import org.apache.fineract.infrastructure.core.domain.FineractContext;
+import org.springframework.batch.integration.partition.StepExecutionRequest;
+
+@Data
+public class COBMessage implements Serializable {
+
+    private StepExecutionRequest stepExecutionRequest;
+    private FineractContext context;
+}

fineract-provider/src/main/java/org/apache/fineract/cob/COBOutputChannelInterceptor.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.cob;
+
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+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 {
+
+    @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);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyService.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;
+
+public interface COBPropertyService {
+
+    Integer getPartitionSize(String jobName);
+
+    Integer getChunkSize(String jobName);
+}

fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyServiceImpl.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;
+
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class COBPropertyServiceImpl implements COBPropertyService {
+
+    private final FineractProperties fineractProperties;
+
+    @Override
+    public Integer getPartitionSize(String jobName) {
+        List<FineractProperties.PartitionedJobProperty> jobProperties = fineractProperties.getPartitionedJob()
+                .getPartitionedJobProperties();
+        return jobProperties.stream() //
+                .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
+                .findFirst() //
+                .map(FineractProperties.PartitionedJobProperty::getPartitionSize) //
+                .orElse(0);
+    }
+
+    @Override
+    public Integer getChunkSize(String jobName) {
+        List<FineractProperties.PartitionedJobProperty> jobProperties = fineractProperties.getPartitionedJob()
+                .getPartitionedJobProperties();
+        return jobProperties.stream() //
+                .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) //
+                .findFirst() //
+                .map(FineractProperties.PartitionedJobProperty::getChunkSize) //
+                .orElse(0);
+    }
+}

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

@@ -0,0 +1,60 @@
+/**
+ * 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 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.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.event.core.MessagingEvent;
+import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
+import org.springframework.integration.handler.LoggingHandler;
+
+@Configuration
+@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
+public class COBWorkerConfig {
+
+    @Bean
+    public IntegrationFlow inboundFlow() {
+        return IntegrationFlows.from(eventListener()) //
+                .log(LoggingHandler.Level.DEBUG) //
+                .channel(inboundRequests()) //
+                .intercept(inputInterceptor()) //
+                .get(); //
+    }
+
+    @Bean
+    public ApplicationEventListeningMessageProducer eventListener() {
+        ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
+        producer.setEventTypes(MessagingEvent.class);
+        return producer;
+    }
+
+    @Bean
+    public QueueChannel inboundRequests() {
+        return new QueueChannel();
+    }
+
+    @Bean
+    public COBInputChannelInterceptor inputInterceptor() {
+        return new COBInputChannelInterceptor();
+    }
+}

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

@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.cob.loan;
+
+import org.apache.fineract.cob.COBPropertyService;
+import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+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.launch.support.RunIdIncrementer;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.DirectChannel;
+
+@Configuration
+@EnableBatchIntegration
+@ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
+public class LoanCOBManagerConfiguration {
+
+    @Autowired
+    private JobBuilderFactory jobBuilderFactory;
+    @Autowired
+    private RemotePartitioningManagerStepBuilderFactory stepBuilderFactory;
+
+    @Autowired
+    private LoanRepository loanRepository;
+    @Autowired
+    private COBPropertyService cobPropertyService;
+    @Autowired
+    private DirectChannel outboundRequests;
+
+    @Bean
+    public LoanCOBPartitioner partitioner() {
+        return new LoanCOBPartitioner(loanRepository, cobPropertyService);
+    }
+
+    @Bean
+    public Step loanCOBStep() {
+        return stepBuilderFactory.get(JobName.LOAN_COB.name()).partitioner("Loan COB worker", partitioner()).outputChannel(outboundRequests)
+                .build();
+    }
+
+    @Bean(name = "loanCOBJob")
+    public Job loanCOBJob() {
+        return jobBuilderFactory.get(JobName.LOAN_COB.name()).start(loanCOBStep()).incrementer(new RunIdIncrementer()).build();
+    }
+
+}

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

@@ -0,0 +1,59 @@
+/**
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import org.apache.fineract.cob.COBPropertyService;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.item.ExecutionContext;
+
+@RequiredArgsConstructor
+public class LoanCOBPartitioner implements Partitioner {
+
+    private final LoanRepository loanRepository;
+    private final COBPropertyService cobPropertyService;
+
+    private static final String PARTITION_PREFIX = "partition";
+    private static final String JOB_NAME = "LOAN_COB";
+
+    @Override
+    public Map<String, ExecutionContext> partition(int gridSize) {
+        int partitionCount = cobPropertyService.getPartitionSize(JOB_NAME);
+        Map<String, ExecutionContext> partitions = new HashMap<>();
+        for (int i = 0; i < partitionCount; i++) {
+            ExecutionContext executionContext = new ExecutionContext();
+            executionContext.put("loanIds", new ArrayList<Integer>());
+            partitions.put(PARTITION_PREFIX + i, executionContext);
+        }
+
+        List<Integer> allNonClosedLoanIds = loanRepository.findAllNonClosedLoanIds();
+        for (int i = 0; i < allNonClosedLoanIds.size(); i++) {
+            String key = PARTITION_PREFIX + (i % partitionCount);
+            ExecutionContext executionContext = partitions.get(key);
+            List<Integer> data = (List<Integer>) executionContext.get("loanIds");
+            data.add(allNonClosedLoanIds.get(i));
+        }
+        return partitions;
+    }
+}

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

@@ -0,0 +1,94 @@
+/**
+ * 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 org.apache.fineract.cob.COBPropertyService;
+import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.portfolio.loanaccount.domain.Loan;
+import org.apache.fineract.portfolio.loanaccount.domain.LoanRepository;
+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.launch.support.RunIdIncrementer;
+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;
+
+@Configuration
+@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
+public class LoanCOBWorkerConfiguration {
+
+    @Autowired
+    private JobBuilderFactory jobBuilderFactory;
+    @Autowired
+    private RemotePartitioningWorkerStepBuilderFactory stepBuilderFactory;
+    @Autowired
+    private COBPropertyService cobPropertyService;
+    @Autowired
+    private LoanRepository loanRepository;
+    @Autowired
+    private QueueChannel inboundRequests;
+
+    @Bean(name = "Loan COB worker")
+    public Step loanCOBWorkerStep() {
+        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
+    public Job loanCOBWorkerJob() {
+        return jobBuilderFactory.get("Loan COB worker").start(loanCOBWorkerStep()).incrementer(new RunIdIncrementer()).build();
+    }
+
+    @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;
+        };
+    }
+
+    @Bean
+    public ItemProcessor<Loan, Loan> itemProcessor() {
+        return null;
+    }
+
+    @Bean
+    public RepositoryItemWriter<Loan> itemWriter() {
+        RepositoryItemWriter<Loan> writer = new RepositoryItemWriter<>();
+        writer.setRepository(loanRepository);
+        writer.setMethodName("save");
+        return writer;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractModeValidationCondition.java

@@ -33,9 +33,11 @@ public class FineractModeValidationCondition implements Condition {
                 .orElse(true);
         boolean isWriteModeEnabled = Optional.ofNullable(context.getEnvironment().getProperty("fineract.mode.write-enabled", Boolean.class))
                 .orElse(true);
-        boolean isBatchModeEnabled = Optional.ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-enabled", Boolean.class))
-                .orElse(true);
-        boolean isValidationFails = !isReadModeEnabled && !isWriteModeEnabled && !isBatchModeEnabled;
+        boolean isBatchManagerModeEnabled = Optional
+                .ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-manager-enabled", Boolean.class)).orElse(true);
+        boolean isBatchWorkerModeEnabled = Optional
+                .ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-worker-enabled", Boolean.class)).orElse(true);
+        boolean isValidationFails = !isReadModeEnabled && !isWriteModeEnabled && !isBatchManagerModeEnabled && !isBatchWorkerModeEnabled;
         if (isValidationFails) {
             log.error(
                     "The Fineract instance type is not configured properly. At least one of these environment variables should be true: FINERACT_MODE_READ_ENABLED, FINERACT_MODE_WRITE_ENABLED, FINERACT_MODE_BATCH_ENABLED");

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java

@@ -39,6 +39,8 @@ public class FineractProperties {
 
     private FineractPartitionedJob partitionedJob;
 
+    private FineractRemoteJobMessageHandlerProperties remoteJobMessageHandler;
+
     @Getter
     @Setter
     public static class FineractTenantProperties {
@@ -93,4 +95,18 @@ public class FineractProperties {
         private Integer partitionSize;
         private Integer threadCount;
     }
+
+    @Getter
+    @Setter
+    public static class FineractRemoteJobMessageHandlerProperties {
+
+        private FineractRemoteJobMessageHandlerSpringEventsProperties springEvents;
+    }
+
+    @Getter
+    @Setter
+    public static class FineractRemoteJobMessageHandlerSpringEventsProperties {
+
+        private boolean enabled;
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractRemoteJobMessageHandlerCondition.java

@@ -0,0 +1,51 @@
+/**
+ * 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.config;
+
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+@Slf4j
+public class FineractRemoteJobMessageHandlerCondition implements Condition {
+
+    @Override
+    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+        boolean isSpringEventsEnabled = Optional
+                .ofNullable(
+                        context.getEnvironment().getProperty("fineract.remote-job-message-handler.spring-events.enabled", Boolean.class))
+                .orElse(true);
+        boolean isBatchManagerModeEnabled = Optional
+                .ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-manager-enabled", Boolean.class)).orElse(true);
+        boolean isBatchWorkerModeEnabled = Optional
+                .ofNullable(context.getEnvironment().getProperty("fineract.mode.batch-worker-enabled", Boolean.class)).orElse(true);
+        // TODO extend this expression with other message handlers in the future
+        boolean isMessageHandlerCountFails = !isSpringEventsEnabled;
+        boolean isSpringEventValidationFails = !(isSpringEventsEnabled && isBatchManagerModeEnabled && isBatchWorkerModeEnabled);
+        if (isMessageHandlerCountFails) {
+            log.error("For remote partitioning jobs exactly one Message Handler must be enabled.");
+        }
+        if (isSpringEventValidationFails) {
+            log.error("If Spring Event Message Handler is enabled, the instance must be Batch Manager and Batch Worker too.");
+        }
+        return isMessageHandlerCountFails || isSpringEventValidationFails;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractValidationCondition.java

@@ -32,4 +32,7 @@ public class FineractValidationCondition extends AnyNestedCondition {
 
     @Conditional(FineractPartitionJobConfigValidationCondition.class)
     static class FineractPartitionedJobValidation {}
+
+    @Conditional(FineractRemoteJobMessageHandlerCondition.class)
+    static class FineractRemoteJobMessageHandlerValidation {}
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobName.java

@@ -54,7 +54,8 @@ public enum JobName {
     UPDATE_TRIAL_BALANCE_DETAILS("Update Trial Balance Details"), //
     EXECUTE_DIRTY_JOBS("Execute All Dirty Jobs"), //
     INCREASE_BUSINESS_DATE_BY_1_DAY("Increase Business Date by 1 day"), //
-    INCREASE_COB_DATE_BY_1_DAY("Increase COB Date by 1 day");
+    INCREASE_COB_DATE_BY_1_DAY("Increase COB Date by 1 day"), //
+    LOAN_COB("Loan COB");
 
     private final String name;
 

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobRegisterServiceImpl.java

@@ -325,7 +325,7 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
         jobDetailFactoryBean.setGroup(scheduledJobDetail.getGroupName());
         jobDetailFactoryBean.setConcurrent(false);
 
-        jobDetailFactoryBean.setArguments(job, scheduledJobDetail);
+        jobDetailFactoryBean.setArguments(job, scheduledJobDetail, ThreadLocalContextUtil.getContext());
         jobDetailFactoryBean.afterPropertiesSet();
         return jobDetailFactoryBean.getObject();
     }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobSchedulerServiceImpl.java

@@ -18,9 +18,13 @@
  */
 package org.apache.fineract.infrastructure.jobs.service;
 
+import java.time.LocalDate;
+import java.util.HashMap;
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
+import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
@@ -40,6 +44,7 @@ public class JobSchedulerServiceImpl implements ApplicationListener<ContextRefre
     private final SchedularWritePlatformService schedularWritePlatformService;
     private final TenantDetailsService tenantDetailsService;
     private final JobRegisterService jobRegisterService;
+    private final BusinessDateReadPlatformService businessDateReadPlatformService;
 
     @Override
     public void onApplicationEvent(ContextRefreshedEvent event) {
@@ -51,6 +56,8 @@ public class JobSchedulerServiceImpl implements ApplicationListener<ContextRefre
         final List<FineractPlatformTenant> allTenants = tenantDetailsService.findAllTenants();
         for (final FineractPlatformTenant tenant : allTenants) {
             ThreadLocalContextUtil.setTenant(tenant);
+            HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
+            ThreadLocalContextUtil.setBusinessDates(businessDates);
             final List<ScheduledJobDetail> scheduledJobDetails = schedularWritePlatformService
                     .retrieveAllJobs(fineractProperties.getNodeId());
             for (final ScheduledJobDetail jobDetails : scheduledJobDetails) {

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java

@@ -22,6 +22,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import lombok.RequiredArgsConstructor;
+import org.apache.fineract.infrastructure.core.domain.FineractContext;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository;
 import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
 import org.springframework.batch.core.Job;
@@ -44,8 +46,10 @@ public class JobStarter {
     private final JobLauncher jobLauncher;
     private final JobParameterRepository jobParameterRepository;
 
-    public void run(Job job, ScheduledJobDetail scheduledJobDetail) throws JobInstanceAlreadyCompleteException,
-            JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException {
+    public void run(Job job, ScheduledJobDetail scheduledJobDetail, FineractContext fineractContext)
+            throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException,
+            JobRestartException {
+        ThreadLocalContextUtil.init(fineractContext);
         Map<String, JobParameter> jobParameterMap = getJobParameter(scheduledJobDetail);
         JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job)
                 .addJobParameters(new JobParameters(jobParameterMap)).toJobParameters();

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobRunnerReadServiceImpl.java

@@ -22,6 +22,7 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Date;
 import java.util.List;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.infrastructure.core.service.Page;
 import org.apache.fineract.infrastructure.core.service.PaginationHelper;
 import org.apache.fineract.infrastructure.core.service.SearchParameters;
@@ -39,6 +40,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 @Service
+@Slf4j
 @Transactional(readOnly = true)
 public class SchedulerJobRunnerReadServiceImpl implements SchedulerJobRunnerReadService {
 

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

@@ -69,6 +69,8 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
 
     String FIND_NON_CLOSED_BY_ACCOUNT_NUMBER = "select loan from Loan loan where loan.accountNumber = :accountNumber and loan.loanStatus in (100,200,300,303,304)";
 
+    String FIND_ALL_NON_CLOSED = "select loan.id from Loan loan where loan.loanStatus in (100,200,300,303,304)";
+
     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";
 
     String FIND_BY_ACCOUNT_NUMBER = "select loan from Loan loan where loan.accountNumber = :accountNumber";
@@ -163,4 +165,7 @@ public interface LoanRepository extends JpaRepository<Loan, Long>, JpaSpecificat
 
     boolean existsByExternalId(@Param("externalId") String externalId);
 
+    @Query(FIND_ALL_NON_CLOSED)
+    List<Integer> findAllNonClosedLoanIds();
+
 }

fineract-provider/src/main/resources/application.properties

@@ -48,6 +48,8 @@ fineract.partitioned-job.partitioned-job-properties[0].chunk-size=${LOAN_COB_CHU
 fineract.partitioned-job.partitioned-job-properties[0].partition-size=${LOAN_COB_PARTITION_SIZE:100}
 fineract.partitioned-job.partitioned-job-properties[0].thread-count=${LOAN_COB_THREAD_COUNT:1}
 
+fineract.remote-job-message-handler.spring-events.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED:true}
+
 # Logging pattern for the console
 logging.pattern.console=${CONSOLE_LOG_PATTERN:%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(%replace([%X{correlationId}]){'\\[\\]', ''}) %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:%wEx}}
 

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

@@ -56,4 +56,5 @@
     <include file="parts/0034_add_audit_entries_to_note.xml" relativeToChangelogFile="true"/>
     <include file="parts/0035_add_audit_entries_to_calendar.xml" relativeToChangelogFile="true"/>
     <include file="parts/0036_add_audit_entries_and_rework_command_source_datetime_fields.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0037_add_loan_cob_job_data.xml" relativeToChangelogFile="true"/>
 </databaseChangeLog>

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

@@ -0,0 +1,45 @@
+<?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">
+        <insert tableName="job">
+            <column name="name" value="Loan COB"/>
+            <column name="display_name" value="Loan COB"/>
+            <column name="cron_expression" value="0 0 0 * * ?"/>
+            <column name="create_time" valueDate="${current_datetime}"/>
+            <column name="task_priority" valueNumeric="5"/>
+            <column name="group_name"/>
+            <column name="previous_run_start_time"/>
+            <column name="job_key" value="Loan COB dayJobDetail1 _ DEFAULT"/>
+            <column name="initializing_errorlog"/>
+            <column name="is_active" valueBoolean="false"/>
+            <column name="currently_running" valueBoolean="false"/>
+            <column name="updates_allowed" valueBoolean="true"/>
+            <column name="scheduler_group" valueNumeric="0"/>
+            <column name="is_misfired" valueBoolean="false"/>
+            <column name="node_id" valueNumeric="1"/>
+            <column name="is_mismatched_job" valueBoolean="false"/>
+        </insert>
+    </changeSet>
+</databaseChangeLog>