galovics commented on code in PR #2491:
URL: https://github.com/apache/fineract/pull/2491#discussion_r946752188
##########
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_MANAGER("Loan COB manager");
Review Comment:
Let's name it simply `Loan COB`. It's a technical detail that a manager is
starting the job.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java:
##########
@@ -0,0 +1,122 @@
+/**
+ * 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.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;
+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;
+
+@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;
+
+ private static final String JOB_NAME = "LOAN_COB";
+
+ @Bean(name = "Loan COB worker")
+ public Step loanCOBWorkerStep() {
+ return stepBuilderFactory.get("Loan COB
worker").inputChannel(inboundRequests())
+ .<Loan,
Loan>chunk(cobPropertyService.getChunkSize(JOB_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['data']}") 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;
+ }
+
+ @Bean
+ public IntegrationFlow inboundFlow() {
+ return IntegrationFlows.from(eventListener()) //
+ .log() //
Review Comment:
I think the log should be configured to DEBUG level otherwise it's too
verbose.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java:
##########
@@ -0,0 +1,122 @@
+/**
+ * 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.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;
+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;
+
+@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;
+
+ private static final String JOB_NAME = "LOAN_COB";
+
+ @Bean(name = "Loan COB worker")
+ public Step loanCOBWorkerStep() {
+ return stepBuilderFactory.get("Loan COB
worker").inputChannel(inboundRequests())
+ .<Loan,
Loan>chunk(cobPropertyService.getChunkSize(JOB_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['data']}") List<Integer> data) {
+ List<Integer> remainingData = new ArrayList<>(data);
+ return () -> {
+ if (remainingData.size() > 0) {
+ return
loanRepository.findById(remainingData.remove(0).longValue()).orElse(null);
Review Comment:
This seems to be an overkill to issue a single statement for every single
ID, don't you think?
We shall rather grab a chunk size of Loans at once.
##########
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";
Review Comment:
The job name should be coming from the JobName enum.
##########
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 {
Review Comment:
I think this could be named as `PartitionedJobPropertyService` cause not
only COBs could be partitioned in the future.
##########
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);
Review Comment:
Doesn't it make sense to accept a JobName enum as an argument here?
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBWorkerConfiguration.java:
##########
@@ -0,0 +1,122 @@
+/**
+ * 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.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;
+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;
+
+@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;
+
+ private static final String JOB_NAME = "LOAN_COB";
+
+ @Bean(name = "Loan COB worker")
+ public Step loanCOBWorkerStep() {
+ return stepBuilderFactory.get("Loan COB
worker").inputChannel(inboundRequests())
+ .<Loan,
Loan>chunk(cobPropertyService.getChunkSize(JOB_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['data']}") List<Integer> data) {
Review Comment:
Can we call the `data` key `loanIds` to be more correct semantically?
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java:
##########
@@ -0,0 +1,91 @@
+/**
+ * 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.batch.integration.partition.StepExecutionRequest;
+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;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import
org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
+import org.springframework.integration.transformer.GenericTransformer;
+
+@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;
+
+ @Bean
+ public LoanCOBPartitioner partitioner() {
+ return new LoanCOBPartitioner(loanRepository, cobPropertyService);
+ }
+
+ @Bean
+ public Step loanCOBStep() {
+ return
stepBuilderFactory.get(JobName.LOAN_COB_MANAGER.name()).partitioner("Loan COB
worker", partitioner())
Review Comment:
Let's separate the Step name from the Job name.
Also, we could make the Step and Job Spring beans having the same names.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBInputChannelTransformer.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.loan;
+
+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 LoanCOBInputChannelTransformer implements
ExecutorChannelInterceptor {
Review Comment:
Rename please.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBManagerConfiguration.java:
##########
@@ -0,0 +1,91 @@
+/**
+ * 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.batch.integration.partition.StepExecutionRequest;
+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;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import
org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
+import org.springframework.integration.transformer.GenericTransformer;
+
+@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;
+
+ @Bean
+ public LoanCOBPartitioner partitioner() {
+ return new LoanCOBPartitioner(loanRepository, cobPropertyService);
+ }
+
+ @Bean
+ public Step loanCOBStep() {
+ return
stepBuilderFactory.get(JobName.LOAN_COB_MANAGER.name()).partitioner("Loan COB
worker", partitioner())
+ .outputChannel(outboundRequests()).build();
+ }
+
+ @Bean(name = "loanCOBJob")
+ public Job loanCOBJob() {
+ return
jobBuilderFactory.get(JobName.LOAN_COB_MANAGER.name()).start(loanCOBStep()).incrementer(new
RunIdIncrementer()).build();
+ }
+
+ @Bean
+ public IntegrationFlow outboundFlow() {
+ ApplicationEventPublishingMessageHandler handler = new
ApplicationEventPublishingMessageHandler();
+ return IntegrationFlows.from(outboundRequests()) //
+ .transform(outputTransformer()) //
+ .log() //
Review Comment:
Log to debug level.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBOutputChannelTransformer.java:
##########
@@ -0,0 +1,34 @@
+/**
+ * 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.infrastructure.core.service.ThreadLocalContextUtil;
+import org.springframework.batch.integration.partition.StepExecutionRequest;
+import org.springframework.integration.transformer.GenericTransformer;
+
+public class LoanCOBOutputChannelTransformer implements
GenericTransformer<StepExecutionRequest, LoanCOBMessage> {
Review Comment:
I think rather than a transformer, this could be an interceptor too, please
check.
##########
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 manager"/>
Review Comment:
Simply Loan COB.
##########
fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/domain/LoanRepository.java:
##########
@@ -163,4 +165,7 @@ List<Loan>
findByGroupOfficeIdsAndLoanStatus(@Param("officeIds") Collection<Long
boolean existsByExternalId(@Param("externalId") String externalId);
+ @Query(FIND_ALL_NON_CLOSED)
+ List<Integer> findAllNonClosedLoan();
Review Comment:
I'd rename this method to `findAllNonClosedLoanIds` since we're returning a
list of IDs, not a list of Loans.
##########
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("data", new ArrayList<Integer>());
+ partitions.put(PARTITION_PREFIX + i, executionContext);
+ }
+
+ List<Integer> allNonClosedLoan = loanRepository.findAllNonClosedLoan();
Review Comment:
I think this should be a paged call, otherwise if we're talking about 2
million loans, it might overflow.
##########
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);
Review Comment:
Aren't we lying a little bit when we say it's a 0 partition? If somebody is
trying to ask for a job that's not remote partitioned, I'd say we shall simply
throw an exception rather and fail fast.
##########
fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanCOBMessage.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.loan;
+
+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 LoanCOBMessage implements Serializable {
Review Comment:
Rename please.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
