Parallelization of Jobs: Parallelizing and paging of recalculate interest for loans (PR #524 for FINERACT-428)

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

@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.infrastructure.jobs.domain;
+
+import org.apache.fineract.infrastructure.core.domain.AbstractPersistableCustom;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+import java.util.Objects;
+
+@Entity
+@Table(name = "job_parameters")
+public class JobParameter extends AbstractPersistableCustom<Long> {
+
+    @Column(name = "job_id", nullable = false)
+    private Long jobId;
+
+    @Column(name = "parameter_name",nullable = true)
+    private String parameterName;
+
+    @Column(name = "parameter_value",nullable = true)
+    private String parameterValue;
+
+
+    public JobParameter(final Long jobId, final String parameterName, final String parameterValue) {
+        this.jobId = jobId;
+        this.parameterName = parameterName;
+        this.parameterValue = parameterValue;
+    }
+
+    public static JobParameter getInstance(final Long jobId, final String parameterName, final String parameterValue){
+        return new JobParameter(jobId,parameterName,parameterValue);
+    }
+
+    public Long getJobId() {
+        return jobId;
+    }
+
+    public void setJobId(final Long jobId) {
+        this.jobId = jobId;
+    }
+
+    public String getParameterName() {
+        return parameterName;
+    }
+
+    public void setParameterName(final String parameterName) {
+        this.parameterName = parameterName;
+    }
+
+    public String getParameterValue() {
+        return parameterValue;
+    }
+
+    public void setParameterValue(final String parameterValue) {
+        this.parameterValue = parameterValue;
+    }
+
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!obj.getClass().equals(getClass())) return false;
+        JobParameter jobParameter = (JobParameter) obj;
+        return jobParameter.getJobId().equals(this.getJobId())
+                && jobParameter.getParameterName().equals(this.getParameterName())
+                && jobParameter.getParameterValue().equals(this.getParameterValue());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(jobId,parameterName,parameterValue);
+    }
+}

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

@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.infrastructure.jobs.domain;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import java.util.List;
+
+public interface JobParameterRepository extends JpaRepository<JobParameter, Long>, JpaSpecificationExecutor<JobParameter> {
+
+    @Query("select jobParameter from JobParameter jobParameter where jobParameter.jobId=:jobId")
+    List<JobParameter> findJobParametersByJobId(@Param("jobId") Long jobId);
+}

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

@@ -18,6 +18,7 @@
  */
 package org.apache.fineract.infrastructure.jobs.service;
 
+
 import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -25,7 +26,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Properties;
 import java.util.TimeZone;
-
+import java.util.Map;
 import javax.annotation.PostConstruct;
 
 import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
@@ -33,6 +34,8 @@ import org.apache.fineract.infrastructure.core.exception.PlatformInternalServerE
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.infrastructure.jobs.annotation.CronMethodParser;
 import org.apache.fineract.infrastructure.jobs.annotation.CronMethodParser.ClassMethodNamesPair;
+import org.apache.fineract.infrastructure.jobs.domain.JobParameter;
+import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository;
 import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
 import org.apache.fineract.infrastructure.jobs.domain.SchedulerDetail;
 import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException;
@@ -78,6 +81,7 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
     private SchedulerJobListener schedulerJobListener;
     private SchedulerStopListener schedulerStopListener;
     private SchedulerTriggerListener globalSchedulerTriggerListener;
+    private JobParameterRepository jobParameterRepository;
 
     private final HashMap<String, Scheduler> schedulers = new HashMap<>(4);
 
@@ -111,6 +115,11 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
         this.globalSchedulerTriggerListener = globalTriggerListener;
     }
 
+    @Autowired
+    public  void setJobParameterRepository(JobParameterRepository jobParameterRepository){
+        this.jobParameterRepository=jobParameterRepository;
+    }
+
     @PostConstruct
     public void loadAllJobs() {
         final List<FineractPlatformTenant> allTenants = this.tenantDetailsService.findAllTenants();
@@ -354,6 +363,14 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
         return jobDetailFactoryBean.getObject();
     }
 
+    public Map<String,String> getJobParameter(ScheduledJobDetail scheduledJobDetail){
+        List<JobParameter> jobParameterList= jobParameterRepository.findJobParametersByJobId(scheduledJobDetail.getId());
+        Map<String,String> jobParameterMap=new HashMap<>();
+        for (JobParameter jobparameter:jobParameterList) {
+            jobParameterMap.put(jobparameter.getParameterName(),jobparameter.getParameterValue());
+        }
+        return  jobParameterMap;
+    }
     private Object getBeanObject(final Class<?> classType) throws ClassNotFoundException {
         final List<Class<?>> typesList = new ArrayList<>();
         final Class<?>[] interfaceType = classType.getInterfaces();

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/service/LoanReadPlatformService.java

@@ -21,6 +21,7 @@ package org.apache.fineract.portfolio.loanaccount.service;
 import java.math.BigDecimal;
 import java.util.Collection;
 import java.util.Date;
+import java.util.List;
 
 import org.apache.fineract.infrastructure.core.service.Page;
 import org.apache.fineract.infrastructure.core.service.SearchParameters;
@@ -111,6 +112,8 @@ public interface LoanReadPlatformService {
 
     Collection<Long> fetchLoansForInterestRecalculation();
 
+    List<Long> fetchLoansForInterestRecalculation(Integer pageSize, Long maxLoanIdInList, String officeHierarchy);
+
     LoanTransactionData retrieveLoanPrePaymentTemplate(Long loanId, LocalDate onDate);
 
     Collection<LoanTransactionData> retrieveWaiverLoanTransactions(Long loanId);
@@ -131,10 +134,13 @@ public interface LoanReadPlatformService {
 
     LoanTransactionData retrieveLoanForeclosureTemplate(final Long loanId, final LocalDate transactionDate);
 
+
 	LoanAccountData retrieveLoanByLoanAccount(String loanAccountNumber);
 	
 	Long retrieveLoanIdByAccountNumber(String loanAccountNumber);
 	
 	String retrieveAccountNumberByAccountId(Long accountId);
 
+    Integer retrieveNumberOfActiveLoans();
+
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/service/LoanReadPlatformServiceImpl.java

@@ -30,6 +30,7 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Collections;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.fineract.accounting.common.AccountingRuleType;
@@ -1844,6 +1845,52 @@ public class LoanReadPlatformServiceImpl implements LoanReadPlatformService {
         }
     }
 
+    @Override
+    public List<Long> fetchLoansForInterestRecalculation(Integer pageSize, Long maxLoanIdInList, String officeHierarchy) {
+        StringBuilder sqlBuilder = new StringBuilder();
+        sqlBuilder.append("SELECT ml.id FROM m_loan ml ");
+        sqlBuilder.append(" left join m_client mc on mc.id = ml.client_id ");
+        sqlBuilder.append(" left join m_office o on mc.office_id = o.id  ");
+        sqlBuilder.append(" INNER JOIN m_loan_repayment_schedule mr on mr.loan_id = ml.id ");
+        sqlBuilder.append(" LEFT JOIN m_loan_disbursement_detail dd on dd.loan_id=ml.id and dd.disbursedon_date is null ");
+        // For Floating rate changes
+        sqlBuilder
+                .append(" left join m_product_loan_floating_rates pfr on ml.product_id = pfr.loan_product_id and ml.is_floating_interest_rate = 1");
+        sqlBuilder.append(" left join m_floating_rates fr on  pfr.floating_rates_id = fr.id");
+        sqlBuilder.append(" left join m_floating_rates_periods frp on fr.id = frp.floating_rates_id ");
+        sqlBuilder.append(" left join m_loan_reschedule_request lrr on lrr.loan_id = ml.id");
+        // this is to identify the applicable rates when base rate is changed
+        sqlBuilder.append(" left join  m_floating_rates bfr on  bfr.is_base_lending_rate = 1");
+        sqlBuilder.append(" left join  m_floating_rates_periods bfrp on  bfr.id = bfrp.floating_rates_id and bfrp.created_date >= ?");
+        sqlBuilder.append(" WHERE ml.loan_status_id = ? ");
+        sqlBuilder.append(" and ml.is_npa = 0 ");
+        sqlBuilder.append(" and ((");
+        sqlBuilder.append("ml.interest_recalculation_enabled = 1 ");
+        sqlBuilder.append(" and (ml.interest_recalcualated_on is null or ml.interest_recalcualated_on <> ?)");
+        sqlBuilder.append(" and ((");
+        sqlBuilder.append(" mr.completed_derived is false ");
+        sqlBuilder.append(" and mr.duedate < ? )");
+        sqlBuilder.append(" or dd.expected_disburse_date < ? )) ");
+        sqlBuilder.append(" or (");
+        sqlBuilder.append(" fr.is_active = 1 and  frp.is_active = 1");
+        sqlBuilder.append(" and (frp.created_date >= ?  or ");
+        sqlBuilder.append("(bfrp.id is not null and frp.is_differential_to_base_lending_rate = 1 and frp.from_date >= bfrp.from_date)) ");
+        sqlBuilder.append("and lrr.loan_id is null");
+        sqlBuilder.append(" ))");
+        sqlBuilder.append(" and ml.id >= ?  and o.hierarchy like ? ");
+        sqlBuilder.append(" group by ml.id ");
+        sqlBuilder.append(" limit ? ");
+        try {
+            String currentdate = formatter.print(DateUtils.getLocalDateOfTenant());
+            // will look only for yesterday modified rates
+            String yesterday = formatter.print(DateUtils.getLocalDateOfTenant().minusDays(1));
+            return Collections.synchronizedList(this.jdbcTemplate.queryForList(sqlBuilder.toString(), Long.class, new Object[] { yesterday,
+                    LoanStatus.ACTIVE.getValue(), currentdate, currentdate, currentdate, yesterday,maxLoanIdInList,officeHierarchy,pageSize }));
+        } catch (final EmptyResultDataAccessException e) {
+            return null;
+        }
+    }
+
     @Override
     public Collection<LoanTransactionData> retrieveWaiverLoanTransactions(final Long loanId) {
         try {
@@ -2210,6 +2257,7 @@ public class LoanReadPlatformServiceImpl implements LoanReadPlatformService {
         }
     }
 
+
 	public String retrieveAccountNumberByAccountId(Long accountId) {
 		try {
 			final String sql = "select loan.account_no from m_loan loan where loan.id = ?";
@@ -2218,4 +2266,12 @@ public class LoanReadPlatformServiceImpl implements LoanReadPlatformService {
 			throw new LoanNotFoundException(accountId);
 		}
 	}
+
+
+    @Override
+    public Integer retrieveNumberOfActiveLoans() {
+        final String sql="select count(*) from m_loan";
+        return this.jdbcTemplate.queryForObject(sql,Integer.class);
     }
+}
+

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/service/LoanSchedularService.java

@@ -19,6 +19,10 @@
 package org.apache.fineract.portfolio.loanaccount.service;
 
 import org.apache.fineract.infrastructure.jobs.exception.JobExecutionException;
+import org.apache.fineract.organisation.office.data.OfficeData;
+import org.apache.fineract.organisation.office.domain.Office;
+
+import java.util.Map;
 
 
 public interface LoanSchedularService {
@@ -27,4 +31,8 @@ public interface LoanSchedularService {
 
     void recalculateInterest() throws JobExecutionException;
 
+    void recalculateInterest(final OfficeData office, final int threadPoolSize, final int batchSize);
+
+    void recalculateInterest(@SuppressWarnings("unused") final Map<String, String> jobParameters);
+
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/service/LoanSchedularServiceImpl.java

@@ -18,12 +18,8 @@
  */
 package org.apache.fineract.portfolio.loanaccount.service;
 
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Collection;
+import java.util.concurrent.*;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
 
 import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
 import org.apache.fineract.infrastructure.core.data.ApiParameterError;
@@ -33,13 +29,18 @@ import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.infrastructure.jobs.annotation.CronTarget;
 import org.apache.fineract.infrastructure.jobs.exception.JobExecutionException;
 import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.organisation.office.data.OfficeData;
+import org.apache.fineract.organisation.office.exception.OfficeNotFoundException;
+import org.apache.fineract.organisation.office.service.OfficeReadPlatformService;
 import org.apache.fineract.portfolio.loanaccount.loanschedule.data.OverdueLoanScheduleData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.dao.CannotAcquireLockException;
 import org.springframework.orm.ObjectOptimisticLockingFailureException;
 import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
 
 @Service
 public class LoanSchedularServiceImpl implements LoanSchedularService {
@@ -48,13 +49,19 @@ public class LoanSchedularServiceImpl implements LoanSchedularService {
 	private final ConfigurationDomainService configurationDomainService;
 	private final LoanReadPlatformService loanReadPlatformService;
 	private final LoanWritePlatformService loanWritePlatformService;
+	private final OfficeReadPlatformService officeReadPlatformService;
+	private final ApplicationContext applicationContext;
 
 	@Autowired
 	public LoanSchedularServiceImpl(final ConfigurationDomainService configurationDomainService,
-            final LoanReadPlatformService loanReadPlatformService, final LoanWritePlatformService loanWritePlatformService) {
+									final LoanReadPlatformService loanReadPlatformService, final LoanWritePlatformService loanWritePlatformService,
+									final OfficeReadPlatformService officeReadPlatformService,
+									final ApplicationContext applicationContext) {
 		this.configurationDomainService = configurationDomainService;
 		this.loanReadPlatformService = loanReadPlatformService;
 		this.loanWritePlatformService = loanWritePlatformService;
+		this.officeReadPlatformService = officeReadPlatformService;
+		this.applicationContext=applicationContext;
 	}
 
 	@Override
@@ -184,4 +191,134 @@ public class LoanSchedularServiceImpl implements LoanSchedularService {
 
 	}
 
+	@Override
+	@CronTarget(jobName = JobName.RECALCULATE_INTEREST_FOR_LOAN)
+	public void recalculateInterest(Map<String, String> jobParameters) {
+		//gets the officeId
+		final String officeId = jobParameters.get("officeId");
+		logger.info(officeId);
+		Long officeIdLong=Long.valueOf(officeId);
+		//gets the Office object
+		final OfficeData office = this.officeReadPlatformService.retrieveOffice(officeIdLong);
+		if(office == null)
+			throw new OfficeNotFoundException(officeIdLong);
+		final int threadPoolSize=Integer.parseInt(jobParameters.get("thread-pool-size"));
+		final int batchSize=Integer.parseInt(jobParameters.get("batch-size"));
+
+		recalculateInterest(office,threadPoolSize,batchSize);
+	}
+
+	@Override
+	public void recalculateInterest(OfficeData office, int threadPoolSize, int batchSize) {
+		final int pageSize = batchSize * threadPoolSize;
+
+		//initialise the executor service with fetched configurations
+		final ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);
+
+		Long maxLoanIdInList = 0L;
+		final String officeHierarchy = office.getHierarchy() + "%";
+
+		//Get the loanIds from service
+		List<Long> loanIds = Collections.synchronizedList(this.loanReadPlatformService
+				.fetchLoansForInterestRecalculation(pageSize, maxLoanIdInList, officeHierarchy));
+
+
+		// gets the loanIds data set iteratively and call addAccuruals for that paginated dataset
+		do {
+			int totalFilteredRecords = loanIds.size();
+			logger.info("Starting accrual - total filtered records - " + totalFilteredRecords);
+			recalculateInterest(loanIds, threadPoolSize, batchSize,
+					executorService);
+			maxLoanIdInList+= pageSize+1;
+			loanIds = Collections.synchronizedList(this.loanReadPlatformService
+					.fetchLoansForInterestRecalculation(pageSize, maxLoanIdInList, officeHierarchy));
+		} while (!CollectionUtils.isEmpty(loanIds));
+
+		//shutdown the executor when done
+		executorService.shutdownNow();
+	}
+
+	private void recalculateInterest(List<Long> loanIds,
+									 int threadPoolSize, int batchSize, final ExecutorService executorService) {
+		StringBuilder sb = new StringBuilder();
+
+		List<Callable<Object>> posters = new ArrayList<Callable<Object>>();
+		int fromIndex = 0;
+		// get the size of current paginated dataset
+		int size = loanIds.size();
+		//calculate the batch size
+		batchSize = (int) Math.ceil(size / threadPoolSize);
+
+		if(batchSize == 0)
+			return;
+
+		int toIndex = (batchSize > size - 1)? size : batchSize ;
+		while(toIndex < size && loanIds.get(toIndex - 1).equals(loanIds.get(toIndex))) {
+			toIndex ++;
+		}
+		boolean lastBatch = false;
+		int loopCount = size/batchSize+1;
+
+		for (long i=0; i < loopCount; i++) {
+			List<Long> subList = safeSubList(loanIds, fromIndex, toIndex);
+			RecalculateInterestPoster poster = (RecalculateInterestPoster) this.applicationContext.getBean("recalculateInterestPoster");
+			poster.setLoanIds(subList);
+			poster.setLoanWritePlatformService(loanWritePlatformService);
+			posters.add(Executors.callable(poster));
+			if(lastBatch)
+				break;
+			if(toIndex + batchSize > size - 1)
+				lastBatch = true;
+			fromIndex = fromIndex + (toIndex - fromIndex);
+			toIndex = (toIndex + batchSize > size - 1)? size : toIndex + batchSize;
+			while(toIndex < size && loanIds.get(toIndex - 1).equals(loanIds.get(toIndex))) {
+				toIndex ++;
+			}
+		}
+
+		try {
+			List<Future<Object>> responses = executorService.invokeAll(posters);
+			checkCompletion(responses);
+		} catch (InterruptedException e1) {
+			logger.error("Interrupted while recalculateInterest", e1);
+		}
+	}
+	//break the lists into sub lists
+	public <T> List<T> safeSubList(List<T> list, int fromIndex, int toIndex) {
+		int size = list.size();
+		if (fromIndex >= size || toIndex <= 0 || fromIndex >= toIndex) {
+			return Collections.emptyList();
+		}
+
+		fromIndex = Math.max(0, fromIndex);
+		toIndex = Math.min(size, toIndex);
+
+		return list.subList(fromIndex, toIndex);
+	}
+	//checks the execution of task by each thread in the executor service
+	private void checkCompletion(List<Future<Object>> responses) {
+		try {
+			for(Future f : responses) {
+				f.get();
+			}
+			boolean allThreadsExecuted = false;
+			int noOfThreadsExecuted = 0;
+			for (Future<Object> future : responses) {
+				if (future.isDone()) {
+					noOfThreadsExecuted++;
+				}
+			}
+			allThreadsExecuted = noOfThreadsExecuted == responses.size();
+			if(!allThreadsExecuted)
+				logger.error("All threads could not execute.");
+		} catch (InterruptedException e1) {
+			logger.error("Interrupted while posting IR entries", e1);
+		}  catch (ExecutionException e2) {
+			logger.error("Execution exception while posting IR entries", e2);
+		}
+	}
+
+
+
+
 }

fineract-provider/src/main/java/org/apache/fineract/portfolio/loanaccount/service/RecalculateInterestPoster.java

@@ -0,0 +1,151 @@
+/**
+ * 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.portfolio.loanaccount.service;
+
+import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
+import org.apache.fineract.infrastructure.configuration.domain.GlobalConfigurationRepository;
+import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
+import org.apache.fineract.infrastructure.core.exception.AbstractPlatformDomainRuleException;
+import org.apache.fineract.infrastructure.core.exception.AbstractPlatformResourceNotFoundException;
+import org.apache.fineract.infrastructure.core.exception.PlatformDataIntegrityException;
+import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
+import org.apache.fineract.infrastructure.jobs.exception.JobExecutionException;
+import org.apache.fineract.portfolio.loanaccount.data.LoanScheduleAccrualData;
+import org.joda.time.LocalDate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.dao.CannotAcquireLockException;
+import org.springframework.orm.ObjectOptimisticLockingFailureException;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.util.*;
+
+
+@Component
+
+@Scope("prototype")
+
+public class RecalculateInterestPoster implements Runnable {
+
+
+
+    private final static Logger logger = LoggerFactory.getLogger(" recalculate interest poster");
+
+    private Collection<Long> loanIds;
+
+    private LoanWritePlatformService loanWritePlatformService;
+
+
+    public void setLoanIds(final Collection<Long> loanIds) {
+
+        this.loanIds = loanIds;
+
+    }
+
+
+
+    public void setLoanWritePlatformService(final LoanWritePlatformService loanWritePlatformService) {
+
+        this.loanWritePlatformService = loanWritePlatformService;
+
+    }
+
+
+
+    @Override
+    public void run() {
+
+        Integer maxNumberOfRetries = ThreadLocalContextUtil.getTenant()
+                .getConnection().getMaxRetriesOnDeadlock();
+        Integer maxIntervalBetweenRetries = ThreadLocalContextUtil.getTenant()
+                .getConnection().getMaxIntervalBetweenRetries();
+
+        int i = 0;
+        if (!loanIds.isEmpty()) {
+            final StringBuilder sb = new StringBuilder();
+            for (Long loanId : loanIds) {
+                logger.info("Loan ID " + loanId);
+                Integer numberOfRetries = 0;
+                while (numberOfRetries <= maxNumberOfRetries) {
+                    try {
+                        this.loanWritePlatformService
+                                .recalculateInterest(loanId);
+                        numberOfRetries = maxNumberOfRetries + 1;
+                    } catch (CannotAcquireLockException
+                            | ObjectOptimisticLockingFailureException exception) {
+                        logger.info("Recalulate interest job has been retried  "
+                                + numberOfRetries + " time(s)");
+                        /***
+                         * Fail if the transaction has been retired for
+                         * maxNumberOfRetries
+                         **/
+                        if (numberOfRetries >= maxNumberOfRetries) {
+                            logger.warn("Recalulate interest job has been retried for the max allowed attempts of "
+                                    + numberOfRetries
+                                    + " and will be rolled back");
+                            sb.append("Recalulate interest job has been retried for the max allowed attempts of "
+                                    + numberOfRetries
+                                    + " and will be rolled back");
+                            break;
+                        }
+                        /***
+                         * Else sleep for a random time (between 1 to 10
+                         * seconds) and continue
+                         **/
+                        try {
+                            Random random = new Random();
+                            int randomNum = random
+                                    .nextInt(maxIntervalBetweenRetries + 1);
+                            Thread.sleep(1000 + (randomNum * 1000));
+                            numberOfRetries = numberOfRetries + 1;
+                        } catch (InterruptedException e) {
+                            sb.append("Interest recalculation for loans failed " + exception.getMessage()) ;
+                            break;
+                        }
+                    } catch (Exception e) {
+                        Throwable realCause = e;
+                        if (e.getCause() != null) {
+                            realCause = e.getCause();
+                        }
+                        logger.error("Interest recalculation for loans failed for account:"	+ loanId + " with message " + realCause.getMessage(), e);
+                        sb.append("Interest recalculation for loans failed for account:").append(loanId).append(" with message ")
+                                .append(realCause.getMessage());
+                        numberOfRetries = maxNumberOfRetries + 1;
+                    }
+                    i++;
+                }
+                logger.info("Loans count " + i);
+            }
+            if (sb.length() > 0) {
+                try {
+                    throw new JobExecutionException(sb.toString());
+                } catch (JobExecutionException e) {
+                    logger.info("JobExecutionException occured :", e);
+                }
+            }
+        }
+
+
+    }
+
+
+
+}

fineract-provider/src/main/resources/sql/migrations/core_db/V344__parallelizing_and_paging_recalculate_interest_for_loans_job.sql

@@ -0,0 +1,33 @@
+--
+-- 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.
+--
+
+CREATE TABLE IF NOT EXISTS `job_parameters` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `job_id` BIGINT(20) NOT NULL,
+  `parameter_name` VARCHAR(100) NOT NULL,
+  `parameter_value` INT(11) NOT NULL,
+  PRIMARY KEY (`id`),
+  KEY `FK_job_id` (`job_id`),
+  CONSTRAINT `FK_job_id` FOREIGN KEY (`job_id`) REFERENCES `job` (`id`)
+
+);
+
+INSERT IGNORE INTO `job_parameters`(`job_id`,`parameter_name`,`parameter_value`) VALUES(17,'thread-pool-size',10);
+INSERT IGNORE INTO `job_parameters`(`job_id`,`parameter_name`,`parameter_value`) VALUES(17,'batch-size',100);
+INSERT IGNORE INTO `job_parameters`(`job_id`,`parameter_name`,`parameter_value`) VALUES(17,'officeId',1);