FINERACT-2191: ThreadLocal context handling during job execution

This commit is contained in:
Oleksii Novikov
2025-03-27 11:44:12 +01:00
committed by Adam Saghy
parent 0f1d53cbdc
commit e584e26c39
20 changed files with 303 additions and 268 deletions
@@ -23,6 +23,7 @@ import java.time.LocalDate;
import java.util.HashMap; import java.util.HashMap;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Builder; import lombok.Builder;
import lombok.EqualsAndHashCode;
import lombok.Getter; import lombok.Getter;
import lombok.extern.jackson.Jacksonized; import lombok.extern.jackson.Jacksonized;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
@@ -31,6 +32,7 @@ import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
@Jacksonized @Jacksonized
@Builder @Builder
@Getter @Getter
@EqualsAndHashCode
public class FineractContext implements Serializable { public class FineractContext implements Serializable {
private final String contextHolder; private final String contextHolder;
@@ -38,4 +40,5 @@ public class FineractContext implements Serializable {
private final String authTokenContext; private final String authTokenContext;
private final HashMap<BusinessDateType, LocalDate> businessDateContext; private final HashMap<BusinessDateType, LocalDate> businessDateContext;
private final ActionContext actionContext; private final ActionContext actionContext;
} }
@@ -20,10 +20,16 @@ package org.apache.fineract.infrastructure.core.domain;
import java.io.Serializable; import java.io.Serializable;
import lombok.Builder; import lombok.Builder;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.jackson.Jacksonized; import lombok.extern.jackson.Jacksonized;
@Jacksonized @Jacksonized
@Builder @Builder
@EqualsAndHashCode
@RequiredArgsConstructor
@Getter
public class FineractPlatformTenant implements Serializable { public class FineractPlatformTenant implements Serializable {
private final Long id; private final Long id;
@@ -32,32 +38,4 @@ public class FineractPlatformTenant implements Serializable {
private final String timezoneId; private final String timezoneId;
private final FineractPlatformTenantConnection connection; 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;
}
} }
@@ -27,7 +27,8 @@ import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
import org.springframework.util.Assert; 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 { public final class ThreadLocalContextUtil {
@@ -124,4 +125,5 @@ public final class ThreadLocalContextUtil {
businessDateContext.remove(); businessDateContext.remove();
actionContext.remove(); actionContext.remove();
} }
} }
@@ -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.business.domain.NoExternalEvent;
import org.apache.fineract.infrastructure.event.external.service.ExternalEventService; import org.apache.fineract.infrastructure.event.external.service.ExternalEventService;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionExecution; import org.springframework.transaction.TransactionExecution;
import org.springframework.transaction.TransactionExecutionListener; import org.springframework.transaction.TransactionExecutionListener;
@@ -198,19 +199,29 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
} }
@Override @Override
public void beforeCommit(TransactionExecution transaction) { public void beforeCommit(@NonNull final TransactionExecution transaction) {
List<BusinessEventWithContext> businessEventWithContexts = transactionBusinessEvents.get().peek(); final List<BusinessEventWithContext> businessEventWithContexts = transactionBusinessEvents.get().peek();
if (!businessEventWithContexts.isEmpty()) { if (businessEventWithContexts.isEmpty()) {
FineractContext originalContext = ThreadLocalContextUtil.getContext(); return;
try {
for (BusinessEventWithContext businessEventWithContext : businessEventWithContexts) {
ThreadLocalContextUtil.init(businessEventWithContext.getFineractContext());
externalEventService.postEvent(businessEventWithContext.getEvent());
}
} finally {
ThreadLocalContextUtil.init(originalContext);
}
} }
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 @Override
@@ -109,7 +109,7 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
eventProducer.sendEvents(partitions); eventProducer.sendEvents(partitions);
} }
private void markEventsAsSent(List<Long> eventIds) { private void markEventsAsSent(final List<Long> eventIds) {
OffsetDateTime sentAt = DateUtils.getAuditOffsetDateTime(); OffsetDateTime sentAt = DateUtils.getAuditOffsetDateTime();
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters // Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
@@ -117,20 +117,23 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize); List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
List<Future<?>> tasks = new ArrayList<>(); List<Future<?>> tasks = new ArrayList<>();
final FineractContext context = ThreadLocalContextUtil.getContext(); final FineractContext context = ThreadLocalContextUtil.getContext();
partitions // partitions.forEach(partitionedEventIds -> {
.forEach(partitionedEventIds -> { tasks.add(threadPoolTaskExecutor.submit(() -> {
tasks.add(threadPoolTaskExecutor.submit(() -> { try {
ThreadLocalContextUtil.init(context); ThreadLocalContextUtil.init(context);
transactionTemplate.execute((status) -> { transactionTemplate.execute((status) -> {
measure(() -> { measure(() -> {
repository.markEventsSent(partitionedEventIds, sentAt); repository.markEventsSent(partitionedEventIds, sentAt);
}, timeTaken -> { }, timeTaken -> {
log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size()); log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
});
return null;
}); });
})); return null;
}); });
} finally {
ThreadLocalContextUtil.reset();
}
}));
});
for (Future<?> task : tasks) { for (Future<?> task : tasks) {
try { try {
task.get(); task.get();
@@ -41,9 +41,12 @@ public class RecalculateInterestPoster implements Callable<Void> {
@Override @Override
public Void call() throws JobExecutionException { public Void call() throws JobExecutionException {
ThreadLocalContextUtil.init(fineractContext); if (loanIds.isEmpty()) {
if (!loanIds.isEmpty()) { return null;
List<Throwable> errors = new ArrayList<>(); }
try {
ThreadLocalContextUtil.init(fineractContext);
final List<Throwable> errors = new ArrayList<>();
for (Long loanId : loanIds) { for (Long loanId : loanIds) {
log.debug("Loan ID {}", loanId); log.debug("Loan ID {}", loanId);
try { try {
@@ -55,6 +58,8 @@ public class RecalculateInterestPoster implements Callable<Void> {
if (!errors.isEmpty()) { if (!errors.isEmpty()) {
throw new JobExecutionException(errors); throw new JobExecutionException(errors);
} }
} finally {
ThreadLocalContextUtil.reset();
} }
return null; return null;
} }
@@ -38,6 +38,9 @@ import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.security.core.context.SecurityContextHolder;
/**
* Tasklet to initialize the thread local context for job execution
*/
@Slf4j @Slf4j
@RequiredArgsConstructor @RequiredArgsConstructor
public class InitialisationTasklet implements Tasklet { public class InitialisationTasklet implements Tasklet {
@@ -51,14 +54,17 @@ public class InitialisationTasklet implements Tasklet {
UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), user.getAuthorities()); UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), user.getAuthorities());
SecurityContextHolder.getContext().setAuthentication(auth); SecurityContextHolder.getContext().setAuthentication(auth);
ThreadLocalContextUtil.setActionContext(ActionContext.COB); ThreadLocalContextUtil.setActionContext(ActionContext.COB);
String businessDateString = Objects.requireNonNull((String) chunkContext.getStepContext().getStepExecution().getJobExecution() String businessDateString = Objects.requireNonNull((String) chunkContext.getStepContext().getStepExecution().getJobExecution()
.getExecutionContext().get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME)); .getExecutionContext().get(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME));
LocalDate businessDate = LocalDate.parse(businessDateString, DateTimeFormatter.ISO_DATE); LocalDate businessDate = LocalDate.parse(businessDateString, DateTimeFormatter.ISO_DATE);
businessDates.put(BusinessDateType.COB_DATE, businessDate); businessDates.put(BusinessDateType.COB_DATE, businessDate);
businessDates.put(BusinessDateType.BUSINESS_DATE, businessDate.plusDays(1)); businessDates.put(BusinessDateType.BUSINESS_DATE, businessDate.plusDays(1));
ThreadLocalContextUtil.setBusinessDates(businessDates); 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; return RepeatStatus.FINISHED;
} }
} }
@@ -18,19 +18,30 @@
*/ */
package org.apache.fineract.cob.loan; 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.domain.FineractContext;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.core.task.TaskDecorator; 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 { public class ContextAwareTaskDecorator implements TaskDecorator {
@NonNull
@Override @Override
public Runnable decorate(@NotNull Runnable runnable) { public Runnable decorate(@NonNull final Runnable runnable) {
final FineractContext context = ThreadLocalContextUtil.getContext(); final FineractContext context = ThreadLocalContextUtil.getContext();
return () -> { return () -> {
ThreadLocalContextUtil.init(context); try {
runnable.run(); log.debug("Initializing thread context for decorated task");
ThreadLocalContextUtil.init(context);
runnable.run();
} finally {
ThreadLocalContextUtil.reset();
}
}; };
} }
@@ -18,7 +18,6 @@
*/ */
package org.apache.fineract.cob.service; package org.apache.fineract.cob.service;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.time.LocalDate; import java.time.LocalDate;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.Collections; 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.DateUtils;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO; 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.ScheduledJobDetail;
import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetailRepository; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetailRepository;
import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException; import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException;
import org.apache.fineract.infrastructure.jobs.service.JobStarter; import org.apache.fineract.infrastructure.jobs.service.JobStarter;
import org.apache.fineract.infrastructure.jobs.service.SchedulerServiceConstants;
import org.quartz.JobExecutionException; import org.quartz.JobExecutionException;
import org.springframework.batch.core.Job; import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParametersInvalidException; import org.springframework.batch.core.JobParametersInvalidException;
@@ -63,12 +62,10 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ
private final JobLocator jobLocator; private final JobLocator jobLocator;
private final ScheduledJobDetailRepository scheduledJobDetailRepository; private final ScheduledJobDetailRepository scheduledJobDetailRepository;
private final JobStarter jobStarter; private final JobStarter jobStarter;
private final JobParameterRepository jobParameterRepository;
private final RetrieveLoanIdService retrieveLoanIdService; private final RetrieveLoanIdService retrieveLoanIdService;
@Override @Override
@Async(TaskExecutorConstant.LOAN_COB_CATCH_UP_TASK_EXECUTOR_BEAN_NAME) @Async(TaskExecutorConstant.LOAN_COB_CATCH_UP_TASK_EXECUTOR_BEAN_NAME)
@SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
public void executeLoanCOBCatchUpAsync(FineractContext context) { public void executeLoanCOBCatchUpAsync(FineractContext context) {
try { try {
ThreadLocalContextUtil.init(context); ThreadLocalContextUtil.init(context);
@@ -80,35 +77,36 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ
? loanIdAndLastClosedBusinessDate.get(0).getLastClosedBusinessDate() ? loanIdAndLastClosedBusinessDate.get(0).getLastClosedBusinessDate()
: cobBusinessDate; : cobBusinessDate;
if (DateUtils.isBefore(oldestCOBProcessedDate, cobBusinessDate)) { if (DateUtils.isBefore(oldestCOBProcessedDate, cobBusinessDate)) {
executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate, context); executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate);
} }
} catch (NoSuchJobException e) { } catch (NoSuchJobException e) {
// Throwing an error here is useless as it will be swallowed hence it is async method // 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 } catch (JobInstanceAlreadyCompleteException | JobRestartException | JobParametersInvalidException
| JobExecutionAlreadyRunningException | JobExecutionException e) { | JobExecutionAlreadyRunningException | JobExecutionException e) {
// Throwing an error here is useless as it will be swallowed hence it is async method // 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 { } finally {
ThreadLocalContextUtil.reset(); ThreadLocalContextUtil.reset();
} }
} }
private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate, private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate)
FineractContext context) throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException,
JobParametersInvalidException, JobRestartException, JobExecutionException { JobParametersInvalidException, JobRestartException, JobExecutionException {
Job job = jobLocator.getJob(LoanCOBConstant.JOB_NAME); Job job = jobLocator.getJob(LoanCOBConstant.JOB_NAME);
ScheduledJobDetail scheduledJobDetail = scheduledJobDetailRepository.findByJobName(LoanCOBConstant.JOB_HUMAN_READABLE_NAME); ScheduledJobDetail scheduledJobDetail = scheduledJobDetailRepository.findByJobName(LoanCOBConstant.JOB_HUMAN_READABLE_NAME);
LocalDate executingBusinessDate = oldestCOBProcessedDate.plusDays(1); LocalDate executingBusinessDate = oldestCOBProcessedDate.plusDays(1);
String tenantIdentifier = ThreadLocalContextUtil.getTenant().getTenantIdentifier();
while (!DateUtils.isAfter(executingBusinessDate, cobBusinessDate)) { 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, JobParameterDTO jobParameterDTO = new JobParameterDTO(LoanCOBConstant.BUSINESS_DATE_PARAMETER_NAME,
executingBusinessDate.format(DateTimeFormatter.ISO_DATE)); executingBusinessDate.format(DateTimeFormatter.ISO_DATE));
JobParameterDTO jobParameterCatchUpDTO = new JobParameterDTO(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, "true"); JobParameterDTO jobParameterCatchUpDTO = new JobParameterDTO(LoanCOBConstant.IS_CATCH_UP_PARAMETER_NAME, "true");
JobParameterDTO tenantParameterDTO = new JobParameterDTO(SchedulerServiceConstants.TENANT_IDENTIFIER, tenantIdentifier);
Set<JobParameterDTO> jobParameters = new HashSet<>(); Set<JobParameterDTO> jobParameters = new HashSet<>();
Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO); Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO, tenantParameterDTO);
jobStarter.run(job, scheduledJobDetail, jobParameters, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); jobStarter.run(job, scheduledJobDetail, jobParameters, tenantIdentifier);
executingBusinessDate = executingBusinessDate.plusDays(1); executingBusinessDate = executingBusinessDate.plusDays(1);
} }
} }
@@ -24,6 +24,8 @@ import java.io.IOException;
import java.net.URLConnection; import java.net.URLConnection;
import java.util.HashSet; import java.util.HashSet;
import java.util.Set; 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.BulkImportEvent;
import org.apache.fineract.infrastructure.bulkimport.data.Count; import org.apache.fineract.infrastructure.bulkimport.data.Count;
import org.apache.fineract.infrastructure.bulkimport.data.GlobalEntityType; 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.exception.GeneralPlatformDomainRuleException;
import org.apache.fineract.infrastructure.core.service.DateUtils; import org.apache.fineract.infrastructure.core.service.DateUtils;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; 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.command.DocumentCommand;
import org.apache.fineract.infrastructure.documentmanagement.domain.Document; import org.apache.fineract.infrastructure.documentmanagement.domain.Document;
import org.apache.fineract.infrastructure.documentmanagement.service.DocumentWritePlatformService; import org.apache.fineract.infrastructure.documentmanagement.service.DocumentWritePlatformService;
import org.apache.poi.ss.usermodel.Workbook; 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.ApplicationContext;
import org.springframework.context.ApplicationListener; import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@Slf4j
@Service @Service
@RequiredArgsConstructor
public class BulkImportEventListener implements ApplicationListener<BulkImportEvent> { public class BulkImportEventListener implements ApplicationListener<BulkImportEvent> {
private static final Logger LOG = LoggerFactory.getLogger(BulkImportEventListener.class);
private final TenantDetailsService tenantDetailsService;
private final ApplicationContext applicationContext; private final ApplicationContext applicationContext;
private final ImportDocumentRepository importRepository; private final ImportDocumentRepository importRepository;
private final DocumentWritePlatformService documentService; 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 @Override
public void onApplicationEvent(final BulkImportEvent event) { 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<String> 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 { 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<String> 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 { try {
workbook.write(bos); try {
} finally { workbook.write(bos);
bos.close(); } finally {
bos.close();
}
} catch (IOException io) {
log.error("Problem occurred in onApplicationEvent function", io);
} }
} catch (IOException io) { byte[] bytes = bos.toByteArray();
LOG.error("Problem occurred in onApplicationEvent function", io); 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);
} }
} }
@@ -42,28 +42,33 @@ public class FineractHookListener implements HookListener {
@Override @Override
public void onApplicationEvent(final HookEvent event) { 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 HookEventSource hookEventSource = (HookEventSource) event.getSource();
final FineractContext fineractContext = event.getContext(); final FineractContext fineractContext = event.getContext();
final String entityName = hookEventSource.getEntityName(); final String entityName = hookEventSource.getEntityName();
final String actionName = hookEventSource.getActionName(); final String actionName = hookEventSource.getActionName();
final String payload = event.getPayload(); final String payload = event.getPayload();
final List<Hook> hooks = hookReadPlatformService.retrieveHooksByEvent(hookEventSource.getEntityName(), final List<Hook> hooks = hookReadPlatformService.retrieveHooksByEvent(hookEventSource.getEntityName(),
hookEventSource.getActionName()); hookEventSource.getActionName());
for (final Hook hook : hooks) { for (final Hook hook : hooks) {
final HookProcessor processor = hookProcessorProvider.getProcessor(hook); final HookProcessor processor = hookProcessorProvider.getProcessor(hook);
try { try {
processor.process(hook, payload, entityName, actionName, fineractContext); processor.process(hook, payload, entityName, actionName, fineractContext);
} catch (Throwable e) { } catch (Throwable e) {
log.error("Hook {} failed in HookProcessor {} for tenantIdentifier/user {}/{}, entityName: {}, actionName: {}, payload {} ", log.error(
hook.getId(), processor.getClass().getSimpleName(), fineractContext.getTenantContext().getTenantIdentifier(), "Hook {} failed in HookProcessor {} for tenantIdentifier/user {}/{}, entityName: {}, actionName: {}, payload {} ",
appUser.getDisplayName(), entityName, actionName, payload, e); hook.getId(), processor.getClass().getSimpleName(), fineractContext.getTenantContext().getTenantIdentifier(),
appUser.getDisplayName(), entityName, actionName, payload, e);
}
} }
} finally {
ThreadLocalContextUtil.reset();
} }
} }
} }
@@ -78,16 +78,24 @@ public class JobStarter {
public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set<JobParameterDTO> jobParameterDTOSet, public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set<JobParameterDTO> jobParameterDTOSet,
String tenantIdentifier) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, String tenantIdentifier) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException,
JobParametersInvalidException, JobRestartException, JobExecutionException { JobParametersInvalidException, JobRestartException, JobExecutionException {
boolean contextInitialized = false;
final FineractPlatformTenant existingTenant = ThreadLocalContextUtil.getTenant();
try { try {
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); if (existingTenant == null) {
ThreadLocalContextUtil.setTenant(tenant); contextInitialized = true;
AppUser user = this.userRepository.fetchSystemUser(); FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), ThreadLocalContextUtil.setTenant(tenant);
user.getAuthorities()); AppUser user = this.userRepository.fetchSystemUser();
SecurityContextHolder.getContext().setAuthentication(auth); UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(),
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates(); user.getAuthorities());
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); SecurityContextHolder.getContext().setAuthentication(auth);
ThreadLocalContextUtil.setBusinessDates(businessDates); HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
ThreadLocalContextUtil.setBusinessDates(businessDates);
}
Map<String, JobParameter<?>> jobParameterMap = getJobParameter(scheduledJobDetail); Map<String, JobParameter<?>> jobParameterMap = getJobParameter(scheduledJobDetail);
JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job) JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job)
.addJobParameters(new JobParameters(jobParameterMap)) .addJobParameters(new JobParameters(jobParameterMap))
@@ -101,7 +109,9 @@ public class JobStarter {
} }
return result; return result;
} finally { } finally {
ThreadLocalContextUtil.reset(); if (contextInitialized) {
ThreadLocalContextUtil.reset();
}
} }
} }
@@ -20,13 +20,11 @@ package org.apache.fineract.infrastructure.jobs.service;
import java.util.Date; import java.util.Date;
import lombok.RequiredArgsConstructor; 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.domain.FineractPlatformTenant;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService; import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService;
import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobRunHistory; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobRunHistory;
import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
import org.quartz.JobExecutionContext; import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException; import org.quartz.JobExecutionException;
import org.quartz.JobKey; import org.quartz.JobKey;
@@ -43,8 +41,6 @@ import org.springframework.stereotype.Component;
public class SchedulerJobListener implements JobListener { public class SchedulerJobListener implements JobListener {
private final SchedularWritePlatformService schedularService; private final SchedularWritePlatformService schedularService;
private final AppUserRepositoryWrapper userRepository;
private final BusinessDateReadPlatformService businessDateReadPlatformService;
private final TenantDetailsService tenantDetailsService; private final TenantDetailsService tenantDetailsService;
private int stackTraceLevel = 0; private int stackTraceLevel = 0;
@@ -61,10 +57,17 @@ public class SchedulerJobListener implements JobListener {
@Override @Override
public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) { 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 { try {
String tenantIdentifier = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); if (existingTenant == null || !existingTenant.getTenantIdentifier().equals(tenantIdentifier)) {
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); contextInitialized = true;
ThreadLocalContextUtil.setTenant(tenant); final FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
ThreadLocalContextUtil.setTenant(tenant);
}
final Trigger trigger = context.getTrigger(); final Trigger trigger = context.getTrigger();
final JobKey key = context.getJobDetail().getKey(); final JobKey key = context.getJobDetail().getKey();
@@ -105,11 +108,12 @@ public class SchedulerJobListener implements JobListener {
final ScheduledJobRunHistory runHistory = new ScheduledJobRunHistory().setScheduledJobDetail(scheduledJobDetails) final ScheduledJobRunHistory runHistory = new ScheduledJobRunHistory().setScheduledJobDetail(scheduledJobDetails)
.setVersion(version).setStartTime(context.getFireTime()).setEndTime(new Date()).setStatus(status) .setVersion(version).setStartTime(context.getFireTime()).setEndTime(new Date()).setStatus(status)
.setErrorMessage(errorMessage).setTriggerType(triggerType).setErrorLog(errorLog); .setErrorMessage(errorMessage).setTriggerType(triggerType).setErrorLog(errorLog);
// scheduledJobDetails.addRunHistory(runHistory);
this.schedularService.saveOrUpdate(scheduledJobDetails, runHistory); this.schedularService.saveOrUpdate(scheduledJobDetails, runHistory);
} finally { } finally {
ThreadLocalContextUtil.reset(); if (contextInitialized) {
ThreadLocalContextUtil.reset();
}
} }
} }
@@ -50,14 +50,22 @@ public class SchedulerTriggerListener implements TriggerListener {
@Override @Override
public boolean vetoJobExecution(final Trigger trigger, final JobExecutionContext context) { public boolean vetoJobExecution(final Trigger trigger, final JobExecutionContext context) {
String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); final String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier); final FineractPlatformTenant existingTenant = ThreadLocalContextUtil.getTenant();
boolean contextInitialized = false;
try { try {
ThreadLocalContextUtil.setTenant(tenant); if (existingTenant == null || !existingTenant.getTenantIdentifier().equals(tenantIdentifier)) {
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); contextInitialized = true;
final FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
ThreadLocalContextUtil.setTenant(tenant);
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
}
return schedulerVetoer.veto(trigger, context); return schedulerVetoer.veto(trigger, context);
} finally { } finally {
ThreadLocalContextUtil.reset(); if (contextInitialized) {
ThreadLocalContextUtil.reset();
}
} }
} }
@@ -25,7 +25,6 @@ import lombok.RequiredArgsConstructor;
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService; import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
import org.apache.fineract.infrastructure.core.domain.ActionContext; 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.JdbcTemplateFactory;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService; 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.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.lang.NonNull;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -55,30 +55,32 @@ public class StuckJobListener implements ApplicationListener<ContextRefreshedEve
private final AppUserRepositoryWrapper userRepository; private final AppUserRepositoryWrapper userRepository;
@Override @Override
public void onApplicationEvent(ContextRefreshedEvent event) { public void onApplicationEvent(@NonNull final ContextRefreshedEvent event) {
if (!jobRegistry.getJobNames().isEmpty()) { if (jobRegistry.getJobNames().isEmpty()) {
List<FineractPlatformTenant> allTenants = tenantDetailsService.findAllTenants(); return;
allTenants.forEach(tenant -> { }
tenantDetailsService.findAllTenants().forEach(tenant -> {
try {
ThreadLocalContextUtil.setTenant(tenant); ThreadLocalContextUtil.setTenant(tenant);
NamedParameterJdbcTemplate namedParameterJdbcTemplate = jdbcTemplateFactory.createNamedParameterJdbcTemplate(tenant); final NamedParameterJdbcTemplate namedParameterJdbcTemplate = jdbcTemplateFactory.createNamedParameterJdbcTemplate(tenant);
List<String> stuckJobNames = jobExecutionRepository.getStuckJobNames(namedParameterJdbcTemplate); final List<String> stuckJobNames = jobExecutionRepository.getStuckJobNames(namedParameterJdbcTemplate);
if (!stuckJobNames.isEmpty()) { if (!stuckJobNames.isEmpty()) {
try { try {
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates(); final HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT); ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
ThreadLocalContextUtil.setBusinessDates(businessDates); ThreadLocalContextUtil.setBusinessDates(businessDates);
AppUser user = userRepository.fetchSystemUser(); final AppUser user = userRepository.fetchSystemUser();
UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), final UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(),
user.getAuthorities()); user.getAuthorities());
SecurityContextHolder.getContext().setAuthentication(auth); SecurityContextHolder.getContext().setAuthentication(auth);
stuckJobNames.forEach(stuckJobExecutorService::resumeStuckJob); stuckJobNames.forEach(stuckJobExecutorService::resumeStuckJob);
} catch (Exception e) {
throw new RuntimeException("Error while trying to restart stuck jobs", e);
} finally { } finally {
ThreadLocalContextUtil.reset(); SecurityContextHolder.getContext().setAuthentication(null);
} }
} }
}); } finally {
} ThreadLocalContextUtil.reset();
}
});
} }
} }
@@ -155,8 +155,12 @@ public class SmsMessageScheduledJobServiceImpl implements SmsMessageScheduledJob
@Override @Override
public void run() { public void run() {
ThreadLocalContextUtil.init(context); try {
connectAndSendToIntermediateServer(apiQueueResourceDatas); ThreadLocalContextUtil.init(context);
connectAndSendToIntermediateServer(apiQueueResourceDatas);
} finally {
ThreadLocalContextUtil.reset();
}
} }
@Override @Override
@@ -18,29 +18,43 @@
*/ */
package org.apache.fineract.infrastructure.springbatch; 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.domain.ActionContext;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.batch.integration.partition.StepExecutionRequest; import org.springframework.batch.integration.partition.StepExecutionRequest;
import org.springframework.lang.NonNull;
import org.springframework.messaging.Message; import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.messaging.support.GenericMessage; 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 { public class InputChannelInterceptor implements ExecutorChannelInterceptor {
@Override @Override
public Message<StepExecutionRequest> beforeHandle(Message<?> message, @NotNull MessageChannel channel, public Message<StepExecutionRequest> beforeHandle(@NonNull final Message<?> message, @NonNull final MessageChannel channel,
@NotNull MessageHandler handler) { @NonNull final MessageHandler handler) {
return beforeHandleMessage(message); 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<StepExecutionRequest> beforeHandleMessage(Message<?> message) { public Message<StepExecutionRequest> beforeHandleMessage(Message<?> message) {
return new GenericMessage<>(beforeHandleMessage((ContextualMessage) message.getPayload())); return new GenericMessage<>(beforeHandleMessage((ContextualMessage) message.getPayload()));
} }
public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) { public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) {
log.debug("Initializing ThreadLocal context for message handling");
ThreadLocalContextUtil.init(contextualMessage.getContext()); ThreadLocalContextUtil.init(contextualMessage.getContext());
ThreadLocalContextUtil.setActionContext(ActionContext.COB); ThreadLocalContextUtil.setActionContext(ActionContext.COB);
return contextualMessage.getStepExecutionRequest(); return contextualMessage.getStepExecutionRequest();
@@ -24,6 +24,7 @@ import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.notification.data.NotificationData; import org.apache.fineract.notification.data.NotificationData;
import org.springframework.context.ApplicationListener; import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.Profile;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@Component @Component
@@ -35,10 +36,15 @@ public class SpringNotificationEventListener implements ApplicationListener<Noti
private final NotificationEventListener notificationEventListener; private final NotificationEventListener notificationEventListener;
@Override @Override
public void onApplicationEvent(NotificationEvent event) { public void onApplicationEvent(@NonNull final NotificationEvent event) {
log.debug("Processing Spring notification event {}", event); log.debug("Processing Spring notification event {}", event);
ThreadLocalContextUtil.init(event.getContext()); try {
NotificationData notificationData = event.getNotificationData(); ThreadLocalContextUtil.init(event.getContext());
notificationEventListener.receive(notificationData); final NotificationData notificationData = event.getNotificationData();
notificationEventListener.receive(notificationData);
} finally {
ThreadLocalContextUtil.reset();
}
} }
} }
@@ -117,23 +117,27 @@ public class PostInterestForSavingTasklet implements Tasklet {
FineractContext context = ThreadLocalContextUtil.getContext(); FineractContext context = ThreadLocalContextUtil.getContext();
Callable<Void> fetchData = () -> { Callable<Void> fetchData = () -> {
ThreadLocalContextUtil.init(context); try {
Long maxId = maxSavingsIdInList; ThreadLocalContextUtil.init(context);
if (!queue.isEmpty()) { Long maxId = maxSavingsIdInList;
maxId = Math.max(maxSavingsIdInList, queue.element().get(queue.element().size() - 1).getId()); 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<SavingsAccountData> 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); while (queue.size() <= QUEUE_SIZE) {
log.debug("Fetching while threads are running!");
List<SavingsAccountData> 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); posters.add(fetchData);
@@ -25,6 +25,7 @@ import lombok.Setter;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.fineract.infrastructure.core.domain.FineractContext; import org.apache.fineract.infrastructure.core.domain.FineractContext;
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil; import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
import org.apache.fineract.infrastructure.jobs.exception.JobExecutionException;
import org.apache.fineract.portfolio.savings.data.SavingsAccountData; import org.apache.fineract.portfolio.savings.data.SavingsAccountData;
/** /**
@@ -40,10 +41,14 @@ public class SavingsSchedularInterestPosterTask implements Callable<Void> {
private FineractContext context; private FineractContext context;
@Override @Override
public Void call() throws org.apache.fineract.infrastructure.jobs.exception.JobExecutionException { public Void call() throws JobExecutionException {
ThreadLocalContextUtil.init(context); try {
interestPoster.postInterest(); ThreadLocalContextUtil.init(context);
return null; interestPoster.postInterest();
return null;
} finally {
ThreadLocalContextUtil.reset();
}
} }
public void setSavingAccounts(Collection<SavingsAccountData> savingAccounts) { public void setSavingAccounts(Collection<SavingsAccountData> savingAccounts) {