From 8a1f1348812bc1b7faae32d814b827be6569a994 Mon Sep 17 00:00:00 2001 From: taskain7 Date: Wed, 3 Aug 2022 19:41:52 +0200 Subject: [PATCH] [FINERACT-1678] Loan COB job skeleton --- fineract-provider/dependencies.gradle | 7 +- .../cob/COBInputChannelInterceptor.java | 36 +++++++ .../apache/fineract/cob/COBManagerConfig.java | 55 +++++++++++ .../org/apache/fineract/cob/COBMessage.java | 31 ++++++ .../cob/COBOutputChannelInterceptor.java | 38 ++++++++ .../fineract/cob/COBPropertyService.java | 26 +++++ .../fineract/cob/COBPropertyServiceImpl.java | 53 +++++++++++ .../apache/fineract/cob/COBWorkerConfig.java | 60 ++++++++++++ .../cob/loan/LoanCOBManagerConfiguration.java | 69 ++++++++++++++ .../fineract/cob/loan/LoanCOBPartitioner.java | 59 ++++++++++++ .../cob/loan/LoanCOBWorkerConfiguration.java | 94 +++++++++++++++++++ .../FineractModeValidationCondition.java | 8 +- .../core/config/FineractProperties.java | 16 ++++ ...eractRemoteJobMessageHandlerCondition.java | 51 ++++++++++ .../config/FineractValidationCondition.java | 3 + .../infrastructure/jobs/service/JobName.java | 3 +- .../jobs/service/JobRegisterServiceImpl.java | 2 +- .../jobs/service/JobSchedulerServiceImpl.java | 7 ++ .../jobs/service/JobStarter.java | 8 +- .../SchedulerJobRunnerReadServiceImpl.java | 2 + .../loanaccount/domain/LoanRepository.java | 5 + .../src/main/resources/application.properties | 2 + .../db/changelog/tenant/changelog-tenant.xml | 1 + .../parts/0037_add_loan_cob_job_data.xml | 45 +++++++++ 24 files changed, 671 insertions(+), 10 deletions(-) create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBInputChannelInterceptor.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBManagerConfig.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBMessage.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBOutputChannelInterceptor.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyService.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyServiceImpl.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/COBWorkerConfig.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBPartitioner.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractRemoteJobMessageHandlerCondition.java create mode 100644 fineract-provider/src/main/resources/db/changelog/tenant/parts/0037_add_loan_cob_job_data.xml diff --git a/fineract-provider/dependencies.gradle b/fineract-provider/dependencies.gradle index 459a89306..ca0992ef3 100644 --- a/fineract-provider/dependencies.gradle +++ b/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' } diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBInputChannelInterceptor.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBInputChannelInterceptor.java new file mode 100644 index 000000000..05626e77b --- /dev/null +++ b/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()); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBManagerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBManagerConfig.java new file mode 100644 index 000000000..9796cc37e --- /dev/null +++ b/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(); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBMessage.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBMessage.java new file mode 100644 index 000000000..f7b04ea7f --- /dev/null +++ b/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; +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBOutputChannelInterceptor.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBOutputChannelInterceptor.java new file mode 100644 index 000000000..5c06c4d94 --- /dev/null +++ b/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); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyService.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyService.java new file mode 100644 index 000000000..58c6d534c --- /dev/null +++ b/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); +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBPropertyServiceImpl.java new file mode 100644 index 000000000..27f3d5a8e --- /dev/null +++ b/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 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 jobProperties = fineractProperties.getPartitionedJob() + .getPartitionedJobProperties(); + return jobProperties.stream() // + .filter(jobProperty -> jobName.equals(jobProperty.getJobName())) // + .findFirst() // + .map(FineractProperties.PartitionedJobProperty::getChunkSize) // + .orElse(0); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/COBWorkerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/cob/COBWorkerConfig.java new file mode 100644 index 000000000..bd8e00043 --- /dev/null +++ b/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(); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java new file mode 100644 index 000000000..077602c74 --- /dev/null +++ b/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(); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBPartitioner.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBPartitioner.java new file mode 100644 index 000000000..49ed65d0b --- /dev/null +++ b/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 partition(int gridSize) { + int partitionCount = cobPropertyService.getPartitionSize(JOB_NAME); + Map partitions = new HashMap<>(); + for (int i = 0; i < partitionCount; i++) { + ExecutionContext executionContext = new ExecutionContext(); + executionContext.put("loanIds", new ArrayList()); + partitions.put(PARTITION_PREFIX + i, executionContext); + } + + List allNonClosedLoanIds = loanRepository.findAllNonClosedLoanIds(); + for (int i = 0; i < allNonClosedLoanIds.size(); i++) { + String key = PARTITION_PREFIX + (i % partitionCount); + ExecutionContext executionContext = partitions.get(key); + List data = (List) executionContext.get("loanIds"); + data.add(allNonClosedLoanIds.get(i)); + } + return partitions; + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java new file mode 100644 index 000000000..18527eb63 --- /dev/null +++ b/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) + .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 itemReader(@Value("#{stepExecutionContext['loanIds']}") List data) { + List remainingData = new ArrayList<>(data); + return () -> { + if (remainingData.size() > 0) { + return loanRepository.findById(remainingData.remove(0).longValue()).orElse(null); + } + return null; + }; + } + + @Bean + public ItemProcessor itemProcessor() { + return null; + } + + @Bean + public RepositoryItemWriter itemWriter() { + RepositoryItemWriter writer = new RepositoryItemWriter<>(); + writer.setRepository(loanRepository); + writer.setMethodName("save"); + return writer; + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractModeValidationCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractModeValidationCondition.java index aff4865a9..13ed4e0ce 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractModeValidationCondition.java +++ b/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"); diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java index e9ff3258e..e90d44dfa 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java +++ b/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; + } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractRemoteJobMessageHandlerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractRemoteJobMessageHandlerCondition.java new file mode 100644 index 000000000..bf57190c0 --- /dev/null +++ b/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; + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractValidationCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractValidationCondition.java index f980ec092..6b8dd5eef 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractValidationCondition.java +++ b/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 {} } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobName.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobName.java index 58e719786..021bbafbe 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobName.java +++ b/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; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobRegisterServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobRegisterServiceImpl.java index fb9e1d8a1..779632ee4 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobRegisterServiceImpl.java +++ b/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(); } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobSchedulerServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobSchedulerServiceImpl.java index 74f13f37d..fa140e7d6 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobSchedulerServiceImpl.java +++ b/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 allTenants = tenantDetailsService.findAllTenants(); for (final FineractPlatformTenant tenant : allTenants) { ThreadLocalContextUtil.setTenant(tenant); + HashMap businessDates = businessDateReadPlatformService.getBusinessDates(); + ThreadLocalContextUtil.setBusinessDates(businessDates); final List scheduledJobDetails = schedularWritePlatformService .retrieveAllJobs(fineractProperties.getNodeId()); for (final ScheduledJobDetail jobDetails : scheduledJobDetails) { diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java index 556e6b4b0..9e72cd54b 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java +++ b/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 jobParameterMap = getJobParameter(scheduledJobDetail); JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job) .addJobParameters(new JobParameters(jobParameterMap)).toJobParameters(); diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobRunnerReadServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobRunnerReadServiceImpl.java index c778f5914..b915521a5 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobRunnerReadServiceImpl.java +++ b/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 { diff --git a/fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java b/fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java index 3ec053973..f75996e70 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java +++ b/fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java @@ -69,6 +69,8 @@ public interface LoanRepository extends JpaRepository, 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, JpaSpecificat boolean existsByExternalId(@Param("externalId") String externalId); + @Query(FIND_ALL_NON_CLOSED) + List findAllNonClosedLoanIds(); + } diff --git a/fineract-provider/src/main/resources/application.properties b/fineract-provider/src/main/resources/application.properties index 89bf99423..f2b4db126 100644 --- a/fineract-provider/src/main/resources/application.properties +++ b/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}} diff --git a/fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml b/fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml index 65cf79c76..501a7aafc 100644 --- a/fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml +++ b/fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml @@ -56,4 +56,5 @@ + diff --git a/fineract-provider/src/main/resources/db/changelog/tenant/parts/0037_add_loan_cob_job_data.xml b/fineract-provider/src/main/resources/db/changelog/tenant/parts/0037_add_loan_cob_job_data.xml new file mode 100644 index 000000000..0b25642bb --- /dev/null +++ b/fineract-provider/src/main/resources/db/changelog/tenant/parts/0037_add_loan_cob_job_data.xml @@ -0,0 +1,45 @@ + + + + + + + + + + + + + + + + + + + + + + + +