diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java index b5e77a276..cd173419b 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java @@ -23,6 +23,7 @@ import java.time.LocalDate; import java.util.HashMap; import lombok.AllArgsConstructor; import lombok.Builder; +import lombok.EqualsAndHashCode; import lombok.Getter; import lombok.extern.jackson.Jacksonized; import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; @@ -31,6 +32,7 @@ import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; @Jacksonized @Builder @Getter +@EqualsAndHashCode public class FineractContext implements Serializable { private final String contextHolder; @@ -38,4 +40,5 @@ public class FineractContext implements Serializable { private final String authTokenContext; private final HashMap businessDateContext; private final ActionContext actionContext; + } diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java index bddce4c68..ed2e81ce3 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java @@ -20,10 +20,16 @@ package org.apache.fineract.infrastructure.core.domain; import java.io.Serializable; import lombok.Builder; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.RequiredArgsConstructor; import lombok.extern.jackson.Jacksonized; @Jacksonized @Builder +@EqualsAndHashCode +@RequiredArgsConstructor +@Getter public class FineractPlatformTenant implements Serializable { private final Long id; @@ -32,32 +38,4 @@ public class FineractPlatformTenant implements Serializable { private final String timezoneId; private final FineractPlatformTenantConnection connection; - public FineractPlatformTenant(final Long id, final String tenantIdentifier, final String name, final String timezoneId, - final FineractPlatformTenantConnection connection) { - this.id = id; - this.tenantIdentifier = tenantIdentifier; - this.name = name; - this.timezoneId = timezoneId; - this.connection = connection; - } - - public Long getId() { - return this.id; - } - - public String getTenantIdentifier() { - return this.tenantIdentifier; - } - - public String getName() { - return this.name; - } - - public String getTimezoneId() { - return this.timezoneId; - } - - public FineractPlatformTenantConnection getConnection() { - return connection; - } } diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/service/ThreadLocalContextUtil.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/service/ThreadLocalContextUtil.java index fe3f10d54..9eb04e491 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/service/ThreadLocalContextUtil.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/service/ThreadLocalContextUtil.java @@ -27,7 +27,8 @@ import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant; import org.springframework.util.Assert; /** - * + * A utility class for managing ThreadLocal context in the application. Provides methods for context initialization and + * cleanup. */ public final class ThreadLocalContextUtil { @@ -124,4 +125,5 @@ public final class ThreadLocalContextUtil { businessDateContext.remove(); actionContext.remove(); } + } diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/business/service/BusinessEventNotifierServiceImpl.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/business/service/BusinessEventNotifierServiceImpl.java index 66eb6824c..4ae4b3702 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/business/service/BusinessEventNotifierServiceImpl.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/business/service/BusinessEventNotifierServiceImpl.java @@ -36,6 +36,7 @@ import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent; import org.apache.fineract.infrastructure.event.business.domain.NoExternalEvent; import org.apache.fineract.infrastructure.event.external.service.ExternalEventService; import org.springframework.beans.factory.InitializingBean; +import org.springframework.lang.NonNull; import org.springframework.stereotype.Service; import org.springframework.transaction.TransactionExecution; import org.springframework.transaction.TransactionExecutionListener; @@ -198,19 +199,29 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe } @Override - public void beforeCommit(TransactionExecution transaction) { - List businessEventWithContexts = transactionBusinessEvents.get().peek(); - if (!businessEventWithContexts.isEmpty()) { - FineractContext originalContext = ThreadLocalContextUtil.getContext(); - try { - for (BusinessEventWithContext businessEventWithContext : businessEventWithContexts) { - ThreadLocalContextUtil.init(businessEventWithContext.getFineractContext()); - externalEventService.postEvent(businessEventWithContext.getEvent()); - } - } finally { - ThreadLocalContextUtil.init(originalContext); - } + public void beforeCommit(@NonNull final TransactionExecution transaction) { + final List businessEventWithContexts = transactionBusinessEvents.get().peek(); + if (businessEventWithContexts.isEmpty()) { + return; } + final FineractContext originalContext = ThreadLocalContextUtil.getContext(); + businessEventWithContexts.forEach(businessEventWithContext -> { + final FineractContext currentContext = businessEventWithContext.getFineractContext(); + boolean swappedContext = false; + try { + if (!originalContext.equals(currentContext)) { + swappedContext = true; + ThreadLocalContextUtil.init(currentContext); + } + externalEventService.postEvent(businessEventWithContext.getEvent()); + } finally { + // Back to original context if we swapped it. We should restore the original context rather than reset + // it completely + if (swappedContext) { + ThreadLocalContextUtil.init(originalContext); + } + } + }); } @Override diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java index 70b514ee8..b52bb661f 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java @@ -109,7 +109,7 @@ public class SendAsynchronousEventsTasklet implements Tasklet { eventProducer.sendEvents(partitions); } - private void markEventsAsSent(List eventIds) { + private void markEventsAsSent(final List eventIds) { OffsetDateTime sentAt = DateUtils.getAuditOffsetDateTime(); // Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters @@ -117,20 +117,23 @@ public class SendAsynchronousEventsTasklet implements Tasklet { List> partitions = Lists.partition(eventIds, partitionSize); List> tasks = new ArrayList<>(); final FineractContext context = ThreadLocalContextUtil.getContext(); - partitions // - .forEach(partitionedEventIds -> { - tasks.add(threadPoolTaskExecutor.submit(() -> { - ThreadLocalContextUtil.init(context); - transactionTemplate.execute((status) -> { - measure(() -> { - repository.markEventsSent(partitionedEventIds, sentAt); - }, timeTaken -> { - log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size()); - }); - return null; + partitions.forEach(partitionedEventIds -> { + tasks.add(threadPoolTaskExecutor.submit(() -> { + try { + ThreadLocalContextUtil.init(context); + transactionTemplate.execute((status) -> { + measure(() -> { + repository.markEventsSent(partitionedEventIds, sentAt); + }, timeTaken -> { + log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size()); }); - })); - }); + return null; + }); + } finally { + ThreadLocalContextUtil.reset(); + } + })); + }); for (Future task : tasks) { try { task.get(); diff --git a/fineract-loan/src/main/java/org/apache/fineract/portfolio/loanaccount/service/RecalculateInterestPoster.java b/fineract-loan/src/main/java/org/apache/fineract/portfolio/loanaccount/service/RecalculateInterestPoster.java index a274a9a98..030e93c29 100644 --- a/fineract-loan/src/main/java/org/apache/fineract/portfolio/loanaccount/service/RecalculateInterestPoster.java +++ b/fineract-loan/src/main/java/org/apache/fineract/portfolio/loanaccount/service/RecalculateInterestPoster.java @@ -41,9 +41,12 @@ public class RecalculateInterestPoster implements Callable { @Override public Void call() throws JobExecutionException { - ThreadLocalContextUtil.init(fineractContext); - if (!loanIds.isEmpty()) { - List errors = new ArrayList<>(); + if (loanIds.isEmpty()) { + return null; + } + try { + ThreadLocalContextUtil.init(fineractContext); + final List errors = new ArrayList<>(); for (Long loanId : loanIds) { log.debug("Loan ID {}", loanId); try { @@ -55,6 +58,8 @@ public class RecalculateInterestPoster implements Callable { if (!errors.isEmpty()) { throw new JobExecutionException(errors); } + } finally { + ThreadLocalContextUtil.reset(); } return null; } diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/common/InitialisationTasklet.java b/fineract-provider/src/main/java/org/apache/fineract/cob/common/InitialisationTasklet.java index 858e999b1..b97d152ca 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/common/InitialisationTasklet.java +++ b/fineract-provider/src/main/java/org/apache/fineract/cob/common/InitialisationTasklet.java @@ -38,6 +38,9 @@ import org.springframework.batch.repeat.RepeatStatus; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; import org.springframework.security.core.context.SecurityContextHolder; +/** + * Tasklet to initialize the thread local context for job execution + */ @Slf4j @RequiredArgsConstructor public class InitialisationTasklet implements Tasklet { @@ -51,14 +54,17 @@ public class InitialisationTasklet implements Tasklet { UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), user.getAuthorities()); SecurityContextHolder.getContext().setAuthentication(auth); ThreadLocalContextUtil.setActionContext(ActionContext.COB); + String businessDateString = Objects.requireNonNull((String) chunkContext.getStepContext().getStepExecution().getJobExecution() .getExecutionContext().get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME)); LocalDate businessDate = LocalDate.parse(businessDateString, DateTimeFormatter.ISO_DATE); + businessDates.put(BusinessDateType.COB_DATE, businessDate); businessDates.put(BusinessDateType.BUSINESS_DATE, businessDate.plusDays(1)); ThreadLocalContextUtil.setBusinessDates(businessDates); - log.debug("Initialisation with Business Date [{}], COB Date [{}] and Action Context [{}]", businessDate.plusDays(1), businessDate, - ThreadLocalContextUtil.getActionContext()); + + log.debug("Initialized context with Business Date [{}], COB Date [{}] and Action Context [{}]", businessDate.plusDays(1), + businessDate, ThreadLocalContextUtil.getActionContext()); return RepeatStatus.FINISHED; } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java index c6bccf0ed..8cde28bbf 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java +++ b/fineract-provider/src/main/java/org/apache/fineract/cob/loan/ContextAwareTaskDecorator.java @@ -18,19 +18,30 @@ */ package org.apache.fineract.cob.loan; +import lombok.extern.slf4j.Slf4j; import org.apache.fineract.infrastructure.core.domain.FineractContext; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; -import org.jetbrains.annotations.NotNull; import org.springframework.core.task.TaskDecorator; +import org.springframework.lang.NonNull; +/** + * Task decorator to ensure proper thread context propagation and cleanup + */ +@Slf4j public class ContextAwareTaskDecorator implements TaskDecorator { + @NonNull @Override - public Runnable decorate(@NotNull Runnable runnable) { + public Runnable decorate(@NonNull final Runnable runnable) { final FineractContext context = ThreadLocalContextUtil.getContext(); return () -> { - ThreadLocalContextUtil.init(context); - runnable.run(); + try { + log.debug("Initializing thread context for decorated task"); + ThreadLocalContextUtil.init(context); + runnable.run(); + } finally { + ThreadLocalContextUtil.reset(); + } }; } diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java index ed6c1428d..92d9c4924 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java +++ b/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java @@ -18,7 +18,6 @@ */ package org.apache.fineract.cob.service; -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.Collections; @@ -37,11 +36,11 @@ import org.apache.fineract.infrastructure.core.domain.FineractContext; import org.apache.fineract.infrastructure.core.service.DateUtils; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO; -import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetailRepository; import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException; import org.apache.fineract.infrastructure.jobs.service.JobStarter; +import org.apache.fineract.infrastructure.jobs.service.SchedulerServiceConstants; import org.quartz.JobExecutionException; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobParametersInvalidException; @@ -63,12 +62,10 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ private final JobLocator jobLocator; private final ScheduledJobDetailRepository scheduledJobDetailRepository; private final JobStarter jobStarter; - private final JobParameterRepository jobParameterRepository; private final RetrieveLoanIdService retrieveLoanIdService; @Override @Async(TaskExecutorConstant.LOAN_COB_CATCH_UP_TASK_EXECUTOR_BEAN_NAME) - @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT") public void executeLoanCOBCatchUpAsync(FineractContext context) { try { ThreadLocalContextUtil.init(context); @@ -80,35 +77,36 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ ? loanIdAndLastClosedBusinessDate.get(0).getLastClosedBusinessDate() : cobBusinessDate; if (DateUtils.isBefore(oldestCOBProcessedDate, cobBusinessDate)) { - executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate, context); + executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate); } } catch (NoSuchJobException e) { // Throwing an error here is useless as it will be swallowed hence it is async method - log.error("", new JobNotFoundException(LoanCOBConstant.JOB_NAME, e)); + log.error("Job not found: {}", LoanCOBConstant.JOB_NAME, new JobNotFoundException(LoanCOBConstant.JOB_NAME, e)); } catch (JobInstanceAlreadyCompleteException | JobRestartException | JobParametersInvalidException | JobExecutionAlreadyRunningException | JobExecutionException e) { // Throwing an error here is useless as it will be swallowed hence it is async method - log.error("", e); + log.error("Error executing job", e); } finally { ThreadLocalContextUtil.reset(); } } - private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate, - FineractContext context) throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, + private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate) + throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException, JobExecutionException { Job job = jobLocator.getJob(LoanCOBConstant.JOB_NAME); ScheduledJobDetail scheduledJobDetail = scheduledJobDetailRepository.findByJobName(LoanCOBConstant.JOB_HUMAN_READABLE_NAME); LocalDate executingBusinessDate = oldestCOBProcessedDate.plusDays(1); + String tenantIdentifier = ThreadLocalContextUtil.getTenant().getTenantIdentifier(); + while (!DateUtils.isAfter(executingBusinessDate, cobBusinessDate)) { - // Need to reinitialize the thread-local tenant info because after running the job, it resets the thread - ThreadLocalContextUtil.init(context); JobParameterDTO jobParameterDTO = new JobParameterDTO(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME, executingBusinessDate.format(DateTimeFormatter.ISO_DATE)); JobParameterDTO jobParameterCatchUpDTO = new JobParameterDTO(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, "true"); + JobParameterDTO tenantParameterDTO = new JobParameterDTO(SchedulerServiceConstants.TENANT_IDENTIFIER, tenantIdentifier); Set jobParameters = new HashSet<>(); - Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO); - jobStarter.run(job, scheduledJobDetail, jobParameters, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); + Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO, tenantParameterDTO); + jobStarter.run(job, scheduledJobDetail, jobParameters, tenantIdentifier); executingBusinessDate = executingBusinessDate.plusDays(1); } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/bulkimport/service/BulkImportEventListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/bulkimport/service/BulkImportEventListener.java index 3c13f6afd..2597c7f78 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/bulkimport/service/BulkImportEventListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/bulkimport/service/BulkImportEventListener.java @@ -24,6 +24,8 @@ import java.io.IOException; import java.net.URLConnection; import java.util.HashSet; import java.util.Set; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.apache.fineract.infrastructure.bulkimport.data.BulkImportEvent; import org.apache.fineract.infrastructure.bulkimport.data.Count; import org.apache.fineract.infrastructure.bulkimport.data.GlobalEntityType; @@ -33,134 +35,88 @@ import org.apache.fineract.infrastructure.bulkimport.importhandler.ImportHandler import org.apache.fineract.infrastructure.core.exception.GeneralPlatformDomainRuleException; import org.apache.fineract.infrastructure.core.service.DateUtils; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; -import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService; import org.apache.fineract.infrastructure.documentmanagement.command.DocumentCommand; import org.apache.fineract.infrastructure.documentmanagement.domain.Document; import org.apache.fineract.infrastructure.documentmanagement.service.DocumentWritePlatformService; import org.apache.poi.ss.usermodel.Workbook; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationListener; import org.springframework.stereotype.Service; +@Slf4j @Service +@RequiredArgsConstructor public class BulkImportEventListener implements ApplicationListener { - private static final Logger LOG = LoggerFactory.getLogger(BulkImportEventListener.class); - private final TenantDetailsService tenantDetailsService; private final ApplicationContext applicationContext; private final ImportDocumentRepository importRepository; private final DocumentWritePlatformService documentService; - @Autowired - public BulkImportEventListener(final TenantDetailsService tenantDetailsService, final ApplicationContext context, - final ImportDocumentRepository importRepository, final DocumentWritePlatformService documentService) { - this.tenantDetailsService = tenantDetailsService; - this.applicationContext = context; - this.importRepository = importRepository; - this.documentService = documentService; - } - @Override public void onApplicationEvent(final BulkImportEvent event) { - ThreadLocalContextUtil.init(event.getContext()); - ImportHandler importHandler = null; - final ImportDocument importDocument = this.importRepository.findById(event.getImportId()).orElse(null); - final GlobalEntityType entityType = GlobalEntityType.fromInt(importDocument.getEntityType()); - - switch (entityType) { - case OFFICES: - importHandler = this.applicationContext.getBean("officeImportHandler", ImportHandler.class); - break; - case CENTERS: - importHandler = this.applicationContext.getBean("centerImportHandler", ImportHandler.class); - break; - case CHART_OF_ACCOUNTS: - importHandler = this.applicationContext.getBean("chartOfAccountsImportHandler", ImportHandler.class); - break; - case CLIENTS_ENTITY: - importHandler = this.applicationContext.getBean("clientEntityImportHandler", ImportHandler.class); - break; - case CLIENTS_PERSON: - importHandler = this.applicationContext.getBean("clientPersonImportHandler", ImportHandler.class); - break; - case FIXED_DEPOSIT_ACCOUNTS: - importHandler = this.applicationContext.getBean("fixedDepositImportHandler", ImportHandler.class); - break; - case FIXED_DEPOSIT_TRANSACTIONS: - importHandler = this.applicationContext.getBean("fixedDepositTransactionImportHandler", ImportHandler.class); - break; - case GROUPS: - importHandler = this.applicationContext.getBean("groupImportHandler", ImportHandler.class); - break; - case GUARANTORS: - importHandler = this.applicationContext.getBean("guarantorImportHandler", ImportHandler.class); - break; - case GL_JOURNAL_ENTRIES: - importHandler = this.applicationContext.getBean("journalEntriesImportHandler", ImportHandler.class); - break; - case LOANS: - importHandler = this.applicationContext.getBean("loanImportHandler", ImportHandler.class); - break; - case LOAN_TRANSACTIONS: - importHandler = this.applicationContext.getBean("loanRepaymentImportHandler", ImportHandler.class); - break; - case RECURRING_DEPOSIT_ACCOUNTS: - importHandler = this.applicationContext.getBean("recurringDepositImportHandler", ImportHandler.class); - break; - case RECURRING_DEPOSIT_ACCOUNTS_TRANSACTIONS: - importHandler = this.applicationContext.getBean("recurringDepositTransactionImportHandler", ImportHandler.class); - break; - case SAVINGS_ACCOUNT: - importHandler = this.applicationContext.getBean("savingsImportHandler", ImportHandler.class); - break; - case SAVINGS_TRANSACTIONS: - importHandler = this.applicationContext.getBean("savingsTransactionImportHandler", ImportHandler.class); - break; - case SHARE_ACCOUNTS: - importHandler = this.applicationContext.getBean("sharedAccountImportHandler", ImportHandler.class); - break; - case STAFF: - importHandler = this.applicationContext.getBean("staffImportHandler", ImportHandler.class); - break; - case USERS: - importHandler = this.applicationContext.getBean("userImportHandler", ImportHandler.class); - break; - default: - throw new GeneralPlatformDomainRuleException("error.msg.unable.to.find.resource", "Unable to find requested resource"); - - } - - final Workbook workbook = event.getWorkbook(); - final Count count = importHandler.process(workbook, event.getLocale(), event.getDateFormat()); - importDocument.update(DateUtils.getLocalDateTimeOfTenant(), count.getSuccessCount(), count.getErrorCount()); - this.importRepository.saveAndFlush(importDocument); - - final Set modifiedParams = new HashSet<>(); - modifiedParams.add("fileName"); - modifiedParams.add("size"); - modifiedParams.add("type"); - modifiedParams.add("location"); - Document document = importDocument.getDocument(); - - DocumentCommand documentCommand = new DocumentCommand(modifiedParams, document.getId(), entityType.name(), null, document.getName(), - document.getFileName(), document.getSize(), URLConnection.guessContentTypeFromName(document.getFileName()), null, null); - - final ByteArrayOutputStream bos = new ByteArrayOutputStream(); try { + ThreadLocalContextUtil.init(event.getContext()); + final ImportDocument importDocument = this.importRepository.findById(event.getImportId()).orElse(null); + final GlobalEntityType entityType = GlobalEntityType.fromInt(importDocument.getEntityType()); + + final ImportHandler importHandler = switch (entityType) { + case OFFICES -> this.applicationContext.getBean("officeImportHandler", ImportHandler.class); + case CENTERS -> this.applicationContext.getBean("centerImportHandler", ImportHandler.class); + case CHART_OF_ACCOUNTS -> this.applicationContext.getBean("chartOfAccountsImportHandler", ImportHandler.class); + case CLIENTS_ENTITY -> this.applicationContext.getBean("clientEntityImportHandler", ImportHandler.class); + case CLIENTS_PERSON -> this.applicationContext.getBean("clientPersonImportHandler", ImportHandler.class); + case FIXED_DEPOSIT_ACCOUNTS -> this.applicationContext.getBean("fixedDepositImportHandler", ImportHandler.class); + case FIXED_DEPOSIT_TRANSACTIONS -> + this.applicationContext.getBean("fixedDepositTransactionImportHandler", ImportHandler.class); + case GROUPS -> this.applicationContext.getBean("groupImportHandler", ImportHandler.class); + case GUARANTORS -> this.applicationContext.getBean("guarantorImportHandler", ImportHandler.class); + case GL_JOURNAL_ENTRIES -> this.applicationContext.getBean("journalEntriesImportHandler", ImportHandler.class); + case LOANS -> this.applicationContext.getBean("loanImportHandler", ImportHandler.class); + case LOAN_TRANSACTIONS -> this.applicationContext.getBean("loanRepaymentImportHandler", ImportHandler.class); + case RECURRING_DEPOSIT_ACCOUNTS -> this.applicationContext.getBean("recurringDepositImportHandler", ImportHandler.class); + case RECURRING_DEPOSIT_ACCOUNTS_TRANSACTIONS -> + this.applicationContext.getBean("recurringDepositTransactionImportHandler", ImportHandler.class); + case SAVINGS_ACCOUNT -> this.applicationContext.getBean("savingsImportHandler", ImportHandler.class); + case SAVINGS_TRANSACTIONS -> this.applicationContext.getBean("savingsTransactionImportHandler", ImportHandler.class); + case SHARE_ACCOUNTS -> this.applicationContext.getBean("sharedAccountImportHandler", ImportHandler.class); + case STAFF -> this.applicationContext.getBean("staffImportHandler", ImportHandler.class); + case USERS -> this.applicationContext.getBean("userImportHandler", ImportHandler.class); + default -> + throw new GeneralPlatformDomainRuleException("error.msg.unable.to.find.resource", "Unable to find requested resource"); + }; + + final Workbook workbook = event.getWorkbook(); + final Count count = importHandler.process(workbook, event.getLocale(), event.getDateFormat()); + importDocument.update(DateUtils.getLocalDateTimeOfTenant(), count.getSuccessCount(), count.getErrorCount()); + this.importRepository.saveAndFlush(importDocument); + + final Set modifiedParams = new HashSet<>(); + modifiedParams.add("fileName"); + modifiedParams.add("size"); + modifiedParams.add("type"); + modifiedParams.add("location"); + Document document = importDocument.getDocument(); + + DocumentCommand documentCommand = new DocumentCommand(modifiedParams, document.getId(), entityType.name(), null, + document.getName(), document.getFileName(), document.getSize(), + URLConnection.guessContentTypeFromName(document.getFileName()), null, null); + + final ByteArrayOutputStream bos = new ByteArrayOutputStream(); try { - workbook.write(bos); - } finally { - bos.close(); + try { + workbook.write(bos); + } finally { + bos.close(); + } + } catch (IOException io) { + log.error("Problem occurred in onApplicationEvent function", io); } - } catch (IOException io) { - LOG.error("Problem occurred in onApplicationEvent function", io); + byte[] bytes = bos.toByteArray(); + ByteArrayInputStream bis = new ByteArrayInputStream(bytes); + this.documentService.updateDocument(documentCommand, bis); + } finally { + ThreadLocalContextUtil.reset(); } - byte[] bytes = bos.toByteArray(); - ByteArrayInputStream bis = new ByteArrayInputStream(bytes); - this.documentService.updateDocument(documentCommand, bis); } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/hooks/listener/FineractHookListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/hooks/listener/FineractHookListener.java index 53e10c4dd..2cfda6434 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/hooks/listener/FineractHookListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/hooks/listener/FineractHookListener.java @@ -42,28 +42,33 @@ public class FineractHookListener implements HookListener { @Override public void onApplicationEvent(final HookEvent event) { - ThreadLocalContextUtil.init(event.getContext()); + try { + ThreadLocalContextUtil.init(event.getContext()); - final AppUser appUser = event.getAppUser(); + final AppUser appUser = event.getAppUser(); - final HookEventSource hookEventSource = (HookEventSource) event.getSource(); - final FineractContext fineractContext = event.getContext(); - final String entityName = hookEventSource.getEntityName(); - final String actionName = hookEventSource.getActionName(); - final String payload = event.getPayload(); + final HookEventSource hookEventSource = (HookEventSource) event.getSource(); + final FineractContext fineractContext = event.getContext(); + final String entityName = hookEventSource.getEntityName(); + final String actionName = hookEventSource.getActionName(); + final String payload = event.getPayload(); - final List hooks = hookReadPlatformService.retrieveHooksByEvent(hookEventSource.getEntityName(), - hookEventSource.getActionName()); + final List hooks = hookReadPlatformService.retrieveHooksByEvent(hookEventSource.getEntityName(), + hookEventSource.getActionName()); - for (final Hook hook : hooks) { - final HookProcessor processor = hookProcessorProvider.getProcessor(hook); - try { - processor.process(hook, payload, entityName, actionName, fineractContext); - } catch (Throwable e) { - log.error("Hook {} failed in HookProcessor {} for tenantIdentifier/user {}/{}, entityName: {}, actionName: {}, payload {} ", - hook.getId(), processor.getClass().getSimpleName(), fineractContext.getTenantContext().getTenantIdentifier(), - appUser.getDisplayName(), entityName, actionName, payload, e); + for (final Hook hook : hooks) { + final HookProcessor processor = hookProcessorProvider.getProcessor(hook); + try { + processor.process(hook, payload, entityName, actionName, fineractContext); + } catch (Throwable e) { + log.error( + "Hook {} failed in HookProcessor {} for tenantIdentifier/user {}/{}, entityName: {}, actionName: {}, payload {} ", + hook.getId(), processor.getClass().getSimpleName(), fineractContext.getTenantContext().getTenantIdentifier(), + appUser.getDisplayName(), entityName, actionName, payload, e); + } } + } finally { + ThreadLocalContextUtil.reset(); } } } 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 a90fb9029..df9175365 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 @@ -78,16 +78,24 @@ public class JobStarter { public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set jobParameterDTOSet, String tenantIdentifier) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException, JobExecutionException { + + boolean contextInitialized = false; + final FineractPlatformTenant existingTenant = ThreadLocalContextUtil.getTenant(); + try { - FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); - ThreadLocalContextUtil.setTenant(tenant); - AppUser user = this.userRepository.fetchSystemUser(); - UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), - user.getAuthorities()); - SecurityContextHolder.getContext().setAuthentication(auth); - HashMap businessDates = businessDateReadPlatformService.getBusinessDates(); - ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); - ThreadLocalContextUtil.setBusinessDates(businessDates); + if (existingTenant == null) { + contextInitialized = true; + FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); + ThreadLocalContextUtil.setTenant(tenant); + AppUser user = this.userRepository.fetchSystemUser(); + UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), + user.getAuthorities()); + SecurityContextHolder.getContext().setAuthentication(auth); + HashMap businessDates = businessDateReadPlatformService.getBusinessDates(); + ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); + ThreadLocalContextUtil.setBusinessDates(businessDates); + } + Map> jobParameterMap = getJobParameter(scheduledJobDetail); JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job) .addJobParameters(new JobParameters(jobParameterMap)) @@ -101,7 +109,9 @@ public class JobStarter { } return result; } finally { - ThreadLocalContextUtil.reset(); + if (contextInitialized) { + ThreadLocalContextUtil.reset(); + } } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobListener.java index 00b169169..a87fc122f 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerJobListener.java @@ -20,13 +20,11 @@ package org.apache.fineract.infrastructure.jobs.service; import java.util.Date; import lombok.RequiredArgsConstructor; -import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService; import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobRunHistory; -import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; import org.quartz.JobKey; @@ -43,8 +41,6 @@ import org.springframework.stereotype.Component; public class SchedulerJobListener implements JobListener { private final SchedularWritePlatformService schedularService; - private final AppUserRepositoryWrapper userRepository; - private final BusinessDateReadPlatformService businessDateReadPlatformService; private final TenantDetailsService tenantDetailsService; private int stackTraceLevel = 0; @@ -61,10 +57,17 @@ public class SchedulerJobListener implements JobListener { @Override public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) { + final String tenantIdentifier = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); + final FineractPlatformTenant existingTenant = ThreadLocalContextUtil.getTenant(); + boolean contextInitialized = false; + try { - String tenantIdentifier = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); - FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); - ThreadLocalContextUtil.setTenant(tenant); + if (existingTenant == null || !existingTenant.getTenantIdentifier().equals(tenantIdentifier)) { + contextInitialized = true; + final FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); + ThreadLocalContextUtil.setTenant(tenant); + } + final Trigger trigger = context.getTrigger(); final JobKey key = context.getJobDetail().getKey(); @@ -105,11 +108,12 @@ public class SchedulerJobListener implements JobListener { final ScheduledJobRunHistory runHistory = new ScheduledJobRunHistory().setScheduledJobDetail(scheduledJobDetails) .setVersion(version).setStartTime(context.getFireTime()).setEndTime(new Date()).setStatus(status) .setErrorMessage(errorMessage).setTriggerType(triggerType).setErrorLog(errorLog); - // scheduledJobDetails.addRunHistory(runHistory); this.schedularService.saveOrUpdate(scheduledJobDetails, runHistory); } finally { - ThreadLocalContextUtil.reset(); + if (contextInitialized) { + ThreadLocalContextUtil.reset(); + } } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerTriggerListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerTriggerListener.java index ead4e9ead..c889bbd94 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerTriggerListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/SchedulerTriggerListener.java @@ -50,14 +50,22 @@ public class SchedulerTriggerListener implements TriggerListener { @Override public boolean vetoJobExecution(final Trigger trigger, final JobExecutionContext context) { - String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); - FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); + final String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); + final FineractPlatformTenant existingTenant = ThreadLocalContextUtil.getTenant(); + boolean contextInitialized = false; + try { - ThreadLocalContextUtil.setTenant(tenant); - ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); + if (existingTenant == null || !existingTenant.getTenantIdentifier().equals(tenantIdentifier)) { + contextInitialized = true; + final FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); + ThreadLocalContextUtil.setTenant(tenant); + ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); + } return schedulerVetoer.veto(trigger, context); } finally { - ThreadLocalContextUtil.reset(); + if (contextInitialized) { + ThreadLocalContextUtil.reset(); + } } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/StuckJobListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/StuckJobListener.java index 21051c12b..c705a5582 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/StuckJobListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/StuckJobListener.java @@ -25,7 +25,6 @@ import lombok.RequiredArgsConstructor; import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService; import org.apache.fineract.infrastructure.core.domain.ActionContext; -import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant; import org.apache.fineract.infrastructure.core.service.JdbcTemplateFactory; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService; @@ -37,6 +36,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.ApplicationListener; import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; +import org.springframework.lang.NonNull; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.stereotype.Service; @@ -55,30 +55,32 @@ public class StuckJobListener implements ApplicationListener allTenants = tenantDetailsService.findAllTenants(); - allTenants.forEach(tenant -> { + public void onApplicationEvent(@NonNull final ContextRefreshedEvent event) { + if (jobRegistry.getJobNames().isEmpty()) { + return; + } + tenantDetailsService.findAllTenants().forEach(tenant -> { + try { ThreadLocalContextUtil.setTenant(tenant); - NamedParameterJdbcTemplate namedParameterJdbcTemplate = jdbcTemplateFactory.createNamedParameterJdbcTemplate(tenant); - List stuckJobNames = jobExecutionRepository.getStuckJobNames(namedParameterJdbcTemplate); + final NamedParameterJdbcTemplate namedParameterJdbcTemplate = jdbcTemplateFactory.createNamedParameterJdbcTemplate(tenant); + final List stuckJobNames = jobExecutionRepository.getStuckJobNames(namedParameterJdbcTemplate); if (!stuckJobNames.isEmpty()) { try { - HashMap businessDates = businessDateReadPlatformService.getBusinessDates(); + final HashMap businessDates = businessDateReadPlatformService.getBusinessDates(); ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); ThreadLocalContextUtil.setBusinessDates(businessDates); - AppUser user = userRepository.fetchSystemUser(); - UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), + final AppUser user = userRepository.fetchSystemUser(); + final UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), user.getAuthorities()); SecurityContextHolder.getContext().setAuthentication(auth); stuckJobNames.forEach(stuckJobExecutorService::resumeStuckJob); - } catch (Exception e) { - throw new RuntimeException("Error while trying to restart stuck jobs", e); } finally { - ThreadLocalContextUtil.reset(); + SecurityContextHolder.getContext().setAuthentication(null); } } - }); - } + } finally { + ThreadLocalContextUtil.reset(); + } + }); } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/sms/scheduler/SmsMessageScheduledJobServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/sms/scheduler/SmsMessageScheduledJobServiceImpl.java index e760ccbd2..278f4fc7b 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/sms/scheduler/SmsMessageScheduledJobServiceImpl.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/sms/scheduler/SmsMessageScheduledJobServiceImpl.java @@ -155,8 +155,12 @@ public class SmsMessageScheduledJobServiceImpl implements SmsMessageScheduledJob @Override public void run() { - ThreadLocalContextUtil.init(context); - connectAndSendToIntermediateServer(apiQueueResourceDatas); + try { + ThreadLocalContextUtil.init(context); + connectAndSendToIntermediateServer(apiQueueResourceDatas); + } finally { + ThreadLocalContextUtil.reset(); + } } @Override diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java index 0d372435a..52ac7be93 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java @@ -18,29 +18,43 @@ */ package org.apache.fineract.infrastructure.springbatch; +import lombok.extern.slf4j.Slf4j; import org.apache.fineract.infrastructure.core.domain.ActionContext; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; -import org.jetbrains.annotations.NotNull; import org.springframework.batch.integration.partition.StepExecutionRequest; +import org.springframework.lang.NonNull; 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; +/** + * Channel interceptor for Spring Batch message handling that ensures ThreadLocal context is properly initialized before + * message handling and cleaned up afterwards + */ +@Slf4j public class InputChannelInterceptor implements ExecutorChannelInterceptor { @Override - public Message beforeHandle(Message message, @NotNull MessageChannel channel, - @NotNull MessageHandler handler) { + public Message beforeHandle(@NonNull final Message message, @NonNull final MessageChannel channel, + @NonNull final MessageHandler handler) { return beforeHandleMessage(message); } + @Override + public void afterMessageHandled(@NonNull final Message message, @NonNull final MessageChannel channel, + @NonNull final MessageHandler handler, final Exception ex) { + log.debug("Cleaning up ThreadLocal context after message handling"); + ThreadLocalContextUtil.reset(); + } + public Message beforeHandleMessage(Message message) { return new GenericMessage<>(beforeHandleMessage((ContextualMessage) message.getPayload())); } public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) { + log.debug("Initializing ThreadLocal context for message handling"); ThreadLocalContextUtil.init(contextualMessage.getContext()); ThreadLocalContextUtil.setActionContext(ActionContext.COB); return contextualMessage.getStepExecutionRequest(); diff --git a/fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/SpringNotificationEventListener.java b/fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/SpringNotificationEventListener.java index 4d427725b..b01903eab 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/SpringNotificationEventListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/notification/eventandlistener/SpringNotificationEventListener.java @@ -24,6 +24,7 @@ import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.notification.data.NotificationData; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Profile; +import org.springframework.lang.NonNull; import org.springframework.stereotype.Component; @Component @@ -35,10 +36,15 @@ public class SpringNotificationEventListener implements ApplicationListener fetchData = () -> { - ThreadLocalContextUtil.init(context); - Long maxId = maxSavingsIdInList; - if (!queue.isEmpty()) { - maxId = Math.max(maxSavingsIdInList, queue.element().get(queue.element().size() - 1).getId()); - } - - while (queue.size() <= QUEUE_SIZE) { - log.debug("Fetching while threads are running!"); - List savingsAccountDataList = Collections.synchronizedList(this.savingAccountReadPlatformService - .retrieveAllSavingsDataForInterestPosting(backdatedTxnsAllowedTill, pageSize, ACTIVE.getValue(), maxId)); - if (savingsAccountDataList.isEmpty()) { - break; + try { + ThreadLocalContextUtil.init(context); + Long maxId = maxSavingsIdInList; + if (!queue.isEmpty()) { + maxId = Math.max(maxSavingsIdInList, queue.element().get(queue.element().size() - 1).getId()); } - maxId = savingsAccountDataList.get(savingsAccountDataList.size() - 1).getId(); - queue.add(savingsAccountDataList); + + while (queue.size() <= QUEUE_SIZE) { + log.debug("Fetching while threads are running!"); + List savingsAccountDataList = Collections.synchronizedList(this.savingAccountReadPlatformService + .retrieveAllSavingsDataForInterestPosting(backdatedTxnsAllowedTill, pageSize, ACTIVE.getValue(), maxId)); + if (savingsAccountDataList.isEmpty()) { + break; + } + maxId = savingsAccountDataList.get(savingsAccountDataList.size() - 1).getId(); + queue.add(savingsAccountDataList); + } + return null; + } finally { + ThreadLocalContextUtil.reset(); } - return null; }; posters.add(fetchData); diff --git a/fineract-savings/src/main/java/org/apache/fineract/portfolio/savings/service/SavingsSchedularInterestPosterTask.java b/fineract-savings/src/main/java/org/apache/fineract/portfolio/savings/service/SavingsSchedularInterestPosterTask.java index 3ebda9bed..2b8fa5a7b 100644 --- a/fineract-savings/src/main/java/org/apache/fineract/portfolio/savings/service/SavingsSchedularInterestPosterTask.java +++ b/fineract-savings/src/main/java/org/apache/fineract/portfolio/savings/service/SavingsSchedularInterestPosterTask.java @@ -25,6 +25,7 @@ import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.fineract.infrastructure.core.domain.FineractContext; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; +import org.apache.fineract.infrastructure.jobs.exception.JobExecutionException; import org.apache.fineract.portfolio.savings.data.SavingsAccountData; /** @@ -40,10 +41,14 @@ public class SavingsSchedularInterestPosterTask implements Callable { private FineractContext context; @Override - public Void call() throws org.apache.fineract.infrastructure.jobs.exception.JobExecutionException { - ThreadLocalContextUtil.init(context); - interestPoster.postInterest(); - return null; + public Void call() throws JobExecutionException { + try { + ThreadLocalContextUtil.init(context); + interestPoster.postInterest(); + return null; + } finally { + ThreadLocalContextUtil.reset(); + } } public void setSavingAccounts(Collection savingAccounts) {