FINERACT-2194: Batch jobs arre failing in a multitenant environment
This commit is contained in:
committed by
Arnold Gálovics
parent
08dc3b123f
commit
c1ec3d7fc7
+61
@@ -0,0 +1,61 @@
|
|||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
|
* or more contributor license agreements. See the NOTICE file
|
||||||
|
* distributed with this work for additional information
|
||||||
|
* regarding copyright ownership. The ASF licenses this file
|
||||||
|
* to you under the Apache License, Version 2.0 (the
|
||||||
|
* "License"); you may not use this file except in compliance
|
||||||
|
* with the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing,
|
||||||
|
* software distributed under the License is distributed on an
|
||||||
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||||
|
* KIND, either express or implied. See the License for the
|
||||||
|
* specific language governing permissions and limitations
|
||||||
|
* under the License.
|
||||||
|
*/
|
||||||
|
package org.apache.fineract.infrastructure.jobs;
|
||||||
|
|
||||||
|
import java.lang.reflect.Method;
|
||||||
|
import java.util.Objects;
|
||||||
|
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
|
||||||
|
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
||||||
|
import org.springframework.cglib.proxy.Factory;
|
||||||
|
import org.springframework.cglib.proxy.MethodInterceptor;
|
||||||
|
import org.springframework.cglib.proxy.MethodProxy;
|
||||||
|
|
||||||
|
public class TenantAwareEqualsHashCodeAdvice implements MethodInterceptor {
|
||||||
|
|
||||||
|
private final Object target;
|
||||||
|
private final String tenantIdentifier;
|
||||||
|
|
||||||
|
public TenantAwareEqualsHashCodeAdvice(Object target) {
|
||||||
|
this.target = target;
|
||||||
|
FineractPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
|
||||||
|
this.tenantIdentifier = tenant != null ? tenant.getTenantIdentifier() : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
|
||||||
|
String methodName = method.getName();
|
||||||
|
|
||||||
|
if ("equals".equals(methodName) && args.length == 1) {
|
||||||
|
Object other = args[0];
|
||||||
|
|
||||||
|
if (other instanceof Factory) {
|
||||||
|
|
||||||
|
TenantAwareEqualsHashCodeAdvice otherProxy = (TenantAwareEqualsHashCodeAdvice) ((Factory) other).getCallback(0);
|
||||||
|
return Objects.equals(target, otherProxy.target) && Objects.equals(tenantIdentifier, otherProxy.tenantIdentifier);
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
if ("hashCode".equals(methodName) && args.length == 0) {
|
||||||
|
return Objects.hash(target.hashCode(), tenantIdentifier);
|
||||||
|
}
|
||||||
|
|
||||||
|
return proxy.invoke(target, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
+67
@@ -0,0 +1,67 @@
|
|||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
|
* or more contributor license agreements. See the NOTICE file
|
||||||
|
* distributed with this work for additional information
|
||||||
|
* regarding copyright ownership. The ASF licenses this file
|
||||||
|
* to you under the Apache License, Version 2.0 (the
|
||||||
|
* "License"); you may not use this file except in compliance
|
||||||
|
* with the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing,
|
||||||
|
* software distributed under the License is distributed on an
|
||||||
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||||
|
* KIND, either express or implied. See the License for the
|
||||||
|
* specific language governing permissions and limitations
|
||||||
|
* under the License.
|
||||||
|
*/
|
||||||
|
package org.springframework.batch.core.scope.context;
|
||||||
|
|
||||||
|
import org.apache.fineract.infrastructure.jobs.TenantAwareEqualsHashCodeAdvice;
|
||||||
|
import org.springframework.batch.core.JobExecution;
|
||||||
|
import org.springframework.batch.core.JobInstance;
|
||||||
|
import org.springframework.batch.core.JobParameters;
|
||||||
|
import org.springframework.cglib.proxy.Enhancer;
|
||||||
|
import org.springframework.lang.Nullable;
|
||||||
|
|
||||||
|
// Temporary solution until spring-batch fixes the concurrency issue
|
||||||
|
// https://github.com/spring-projects/spring-batch/issues/4774
|
||||||
|
// Mostly copy from spring-batch
|
||||||
|
@SuppressWarnings({ "HideUtilityClassConstructor" })
|
||||||
|
public class JobSynchronizationManager {
|
||||||
|
|
||||||
|
private static final SynchronizationManagerSupport<JobExecution, JobContext> manager = new SynchronizationManagerSupport<>() {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected JobContext createNewContext(JobExecution execution) {
|
||||||
|
return new JobContext(execution);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void close(JobContext context) {
|
||||||
|
context.close();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
@Nullable
|
||||||
|
public static JobContext getContext() {
|
||||||
|
return manager.getContext();
|
||||||
|
}
|
||||||
|
|
||||||
|
public static JobContext register(JobExecution jobExecution) {
|
||||||
|
Enhancer enhancer = new Enhancer();
|
||||||
|
enhancer.setSuperclass(JobExecution.class);
|
||||||
|
enhancer.setCallback(new TenantAwareEqualsHashCodeAdvice(jobExecution));
|
||||||
|
return manager.register((JobExecution) enhancer.create(new Class[] { JobInstance.class, Long.class, JobParameters.class },
|
||||||
|
new Object[] { jobExecution.getJobInstance(), jobExecution.getId(), jobExecution.getJobParameters() }));
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void close() {
|
||||||
|
manager.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void release() {
|
||||||
|
manager.release();
|
||||||
|
}
|
||||||
|
}
|
||||||
+67
@@ -0,0 +1,67 @@
|
|||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
|
* or more contributor license agreements. See the NOTICE file
|
||||||
|
* distributed with this work for additional information
|
||||||
|
* regarding copyright ownership. The ASF licenses this file
|
||||||
|
* to you under the Apache License, Version 2.0 (the
|
||||||
|
* "License"); you may not use this file except in compliance
|
||||||
|
* with the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing,
|
||||||
|
* software distributed under the License is distributed on an
|
||||||
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||||
|
* KIND, either express or implied. See the License for the
|
||||||
|
* specific language governing permissions and limitations
|
||||||
|
* under the License.
|
||||||
|
*/
|
||||||
|
package org.springframework.batch.core.scope.context;
|
||||||
|
|
||||||
|
import org.apache.fineract.infrastructure.jobs.TenantAwareEqualsHashCodeAdvice;
|
||||||
|
import org.springframework.batch.core.JobExecution;
|
||||||
|
import org.springframework.batch.core.StepExecution;
|
||||||
|
import org.springframework.cglib.proxy.Enhancer;
|
||||||
|
import org.springframework.lang.Nullable;
|
||||||
|
|
||||||
|
// Temporary solution until spring-batch fixes the concurrency issue
|
||||||
|
// https://github.com/spring-projects/spring-batch/issues/4774
|
||||||
|
// Mostly copy from spring-batch
|
||||||
|
@SuppressWarnings({ "HideUtilityClassConstructor" })
|
||||||
|
public class StepSynchronizationManager {
|
||||||
|
|
||||||
|
private static final SynchronizationManagerSupport<StepExecution, StepContext> manager = new SynchronizationManagerSupport<>() {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected StepContext createNewContext(StepExecution execution) {
|
||||||
|
return new StepContext(execution);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected void close(StepContext context) {
|
||||||
|
context.close();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
@Nullable
|
||||||
|
public static StepContext getContext() {
|
||||||
|
return manager.getContext();
|
||||||
|
}
|
||||||
|
|
||||||
|
public static StepContext register(StepExecution stepExecution) {
|
||||||
|
Enhancer enhancer = new Enhancer();
|
||||||
|
enhancer.setSuperclass(StepExecution.class);
|
||||||
|
enhancer.setCallback(new TenantAwareEqualsHashCodeAdvice(stepExecution));
|
||||||
|
return manager.register((StepExecution) enhancer.create(new Class[] { String.class, JobExecution.class },
|
||||||
|
new Object[] { stepExecution.getStepName(), stepExecution.getJobExecution() }));
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void close() {
|
||||||
|
manager.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void release() {
|
||||||
|
manager.release();
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
+6
-4
@@ -80,7 +80,7 @@ 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);
|
executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate, context);
|
||||||
}
|
}
|
||||||
} 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
|
||||||
@@ -94,19 +94,21 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate)
|
private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate,
|
||||||
throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException,
|
FineractContext context) 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);
|
||||||
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");
|
||||||
Set<JobParameterDTO> jobParameters = new HashSet<>();
|
Set<JobParameterDTO> jobParameters = new HashSet<>();
|
||||||
Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO);
|
Collections.addAll(jobParameters, jobParameterDTO, jobParameterCatchUpDTO);
|
||||||
jobStarter.run(job, scheduledJobDetail, jobParameters);
|
jobStarter.run(job, scheduledJobDetail, jobParameters, ThreadLocalContextUtil.getTenant().getTenantIdentifier());
|
||||||
executingBusinessDate = executingBusinessDate.plusDays(1);
|
executingBusinessDate = executingBusinessDate.plusDays(1);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -28,7 +28,7 @@ public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandle
|
|||||||
@Override
|
@Override
|
||||||
public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
|
public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
|
||||||
|
|
||||||
log.error("Exception message - {}", throwable.getMessage());
|
log.error("Exception", throwable);
|
||||||
log.error("Method name - {}", method.getName());
|
log.error("Method name - {}", method.getName());
|
||||||
for (Object param : obj) {
|
for (Object param : obj) {
|
||||||
log.error("Parameter value - {}", param);
|
log.error("Parameter value - {}", param);
|
||||||
|
|||||||
+1
-1
@@ -349,7 +349,7 @@ public class JobRegisterServiceImpl implements JobRegisterService, ApplicationLi
|
|||||||
jobDetailFactoryBean.setGroup(scheduledJobDetail.getGroupName());
|
jobDetailFactoryBean.setGroup(scheduledJobDetail.getGroupName());
|
||||||
jobDetailFactoryBean.setConcurrent(false);
|
jobDetailFactoryBean.setConcurrent(false);
|
||||||
|
|
||||||
jobDetailFactoryBean.setArguments(job, scheduledJobDetail, jobParameterDTOSet);
|
jobDetailFactoryBean.setArguments(job, scheduledJobDetail, jobParameterDTOSet, tenant.getTenantIdentifier());
|
||||||
jobDetailFactoryBean.afterPropertiesSet();
|
jobDetailFactoryBean.afterPropertiesSet();
|
||||||
return jobDetailFactoryBean.getObject();
|
return jobDetailFactoryBean.getObject();
|
||||||
}
|
}
|
||||||
|
|||||||
+41
-13
@@ -18,6 +18,7 @@
|
|||||||
*/
|
*/
|
||||||
package org.apache.fineract.infrastructure.jobs.service;
|
package org.apache.fineract.infrastructure.jobs.service;
|
||||||
|
|
||||||
|
import java.time.LocalDate;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@@ -27,11 +28,19 @@ import java.util.Set;
|
|||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
|
||||||
|
import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
|
||||||
|
import org.apache.fineract.infrastructure.core.domain.ActionContext;
|
||||||
|
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.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.JobParameterRepository;
|
||||||
import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
|
import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService;
|
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService;
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider;
|
import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider;
|
||||||
|
import org.apache.fineract.useradministration.domain.AppUser;
|
||||||
|
import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
|
||||||
import org.quartz.JobExecutionException;
|
import org.quartz.JobExecutionException;
|
||||||
import org.springframework.batch.core.BatchStatus;
|
import org.springframework.batch.core.BatchStatus;
|
||||||
import org.springframework.batch.core.Job;
|
import org.springframework.batch.core.Job;
|
||||||
@@ -45,6 +54,8 @@ import org.springframework.batch.core.launch.JobLauncher;
|
|||||||
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
|
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
|
||||||
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
|
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
|
||||||
import org.springframework.batch.core.repository.JobRestartException;
|
import org.springframework.batch.core.repository.JobRestartException;
|
||||||
|
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
|
||||||
|
import org.springframework.security.core.context.SecurityContextHolder;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
@Component
|
@Component
|
||||||
@@ -57,24 +68,41 @@ public class JobStarter {
|
|||||||
private final JobParameterRepository jobParameterRepository;
|
private final JobParameterRepository jobParameterRepository;
|
||||||
private final List<JobParameterProvider<?>> jobParameterProviders;
|
private final List<JobParameterProvider<?>> jobParameterProviders;
|
||||||
private final JobNameService jobNameService;
|
private final JobNameService jobNameService;
|
||||||
|
private final TenantDetailsService tenantDetailsService;
|
||||||
|
private final AppUserRepositoryWrapper userRepository;
|
||||||
|
private final BusinessDateReadPlatformService businessDateReadPlatformService;
|
||||||
|
|
||||||
public static final List<BatchStatus> FAILED_STATUSES = List.of(BatchStatus.FAILED, BatchStatus.ABANDONED, BatchStatus.STOPPED,
|
public static final List<BatchStatus> FAILED_STATUSES = List.of(BatchStatus.FAILED, BatchStatus.ABANDONED, BatchStatus.STOPPED,
|
||||||
BatchStatus.STOPPING, BatchStatus.UNKNOWN);
|
BatchStatus.STOPPING, BatchStatus.UNKNOWN);
|
||||||
|
|
||||||
public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set<JobParameterDTO> jobParameterDTOSet)
|
public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set<JobParameterDTO> jobParameterDTOSet,
|
||||||
throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException,
|
String tenantIdentifier) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException,
|
||||||
JobRestartException, JobExecutionException {
|
JobParametersInvalidException, JobRestartException, JobExecutionException {
|
||||||
Map<String, JobParameter<?>> jobParameterMap = getJobParameter(scheduledJobDetail);
|
try {
|
||||||
JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job)
|
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
|
||||||
.addJobParameters(new JobParameters(jobParameterMap))
|
ThreadLocalContextUtil.setTenant(tenant);
|
||||||
.addJobParameters(new JobParameters(provideCustomJobParameters(
|
AppUser user = this.userRepository.fetchSystemUser();
|
||||||
jobNameService.getJobByHumanReadableName(scheduledJobDetail.getJobName()).getEnumStyleName(), jobParameterDTOSet)))
|
UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(),
|
||||||
.toJobParameters();
|
user.getAuthorities());
|
||||||
JobExecution result = jobLauncher.run(job, jobParameters);
|
SecurityContextHolder.getContext().setAuthentication(auth);
|
||||||
if (FAILED_STATUSES.contains(result.getStatus())) {
|
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
|
||||||
throw new JobExecutionException(result.getExitStatus().toString());
|
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
||||||
|
ThreadLocalContextUtil.setBusinessDates(businessDates);
|
||||||
|
Map<String, JobParameter<?>> jobParameterMap = getJobParameter(scheduledJobDetail);
|
||||||
|
JobParameters jobParameters = new JobParametersBuilder(jobExplorer).getNextJobParameters(job)
|
||||||
|
.addJobParameters(new JobParameters(jobParameterMap))
|
||||||
|
.addJobParameters(new JobParameters(provideCustomJobParameters(
|
||||||
|
jobNameService.getJobByHumanReadableName(scheduledJobDetail.getJobName()).getEnumStyleName(),
|
||||||
|
jobParameterDTOSet)))
|
||||||
|
.toJobParameters();
|
||||||
|
JobExecution result = jobLauncher.run(job, jobParameters);
|
||||||
|
if (FAILED_STATUSES.contains(result.getStatus())) {
|
||||||
|
throw new JobExecutionException(result.getExitStatus().toString());
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
} finally {
|
||||||
|
ThreadLocalContextUtil.reset();
|
||||||
}
|
}
|
||||||
return result;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
protected Map<String, org.springframework.batch.core.JobParameter<?>> getJobParameter(ScheduledJobDetail scheduledJobDetail) {
|
protected Map<String, org.springframework.batch.core.JobParameter<?>> getJobParameter(ScheduledJobDetail scheduledJobDetail) {
|
||||||
|
|||||||
+52
-58
@@ -18,25 +18,20 @@
|
|||||||
*/
|
*/
|
||||||
package org.apache.fineract.infrastructure.jobs.service;
|
package org.apache.fineract.infrastructure.jobs.service;
|
||||||
|
|
||||||
import java.time.LocalDate;
|
|
||||||
import java.util.Date;
|
import java.util.Date;
|
||||||
import java.util.HashMap;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
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.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.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.AppUser;
|
|
||||||
import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
|
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;
|
||||||
import org.quartz.JobListener;
|
import org.quartz.JobListener;
|
||||||
import org.quartz.Trigger;
|
import org.quartz.Trigger;
|
||||||
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
|
|
||||||
import org.springframework.security.core.context.SecurityContextHolder;
|
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -50,6 +45,7 @@ public class SchedulerJobListener implements JobListener {
|
|||||||
private final SchedularWritePlatformService schedularService;
|
private final SchedularWritePlatformService schedularService;
|
||||||
private final AppUserRepositoryWrapper userRepository;
|
private final AppUserRepositoryWrapper userRepository;
|
||||||
private final BusinessDateReadPlatformService businessDateReadPlatformService;
|
private final BusinessDateReadPlatformService businessDateReadPlatformService;
|
||||||
|
private final TenantDetailsService tenantDetailsService;
|
||||||
private int stackTraceLevel = 0;
|
private int stackTraceLevel = 0;
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@@ -58,65 +54,63 @@ public class SchedulerJobListener implements JobListener {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void jobToBeExecuted(@SuppressWarnings("unused") final JobExecutionContext context) {
|
public void jobToBeExecuted(@SuppressWarnings("unused") final JobExecutionContext context) {}
|
||||||
AppUser user = this.userRepository.fetchSystemUser();
|
|
||||||
UsernamePasswordAuthenticationToken auth = new UsernamePasswordAuthenticationToken(user, user.getPassword(), user.getAuthorities());
|
|
||||||
SecurityContextHolder.getContext().setAuthentication(auth);
|
|
||||||
HashMap<BusinessDateType, LocalDate> businessDates = businessDateReadPlatformService.getBusinessDates();
|
|
||||||
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
|
||||||
ThreadLocalContextUtil.setBusinessDates(businessDates);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void jobExecutionVetoed(@SuppressWarnings("unused") final JobExecutionContext context) {
|
public void jobExecutionVetoed(@SuppressWarnings("unused") final JobExecutionContext context) {}
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) {
|
public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) {
|
||||||
final Trigger trigger = context.getTrigger();
|
try {
|
||||||
final JobKey key = context.getJobDetail().getKey();
|
String tenantIdentifier = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
|
||||||
final String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
|
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
|
||||||
final ScheduledJobDetail scheduledJobDetails = this.schedularService.findByJobKey(jobKey);
|
ThreadLocalContextUtil.setTenant(tenant);
|
||||||
final Long version = this.schedularService.fetchMaxVersionBy(jobKey) + 1;
|
final Trigger trigger = context.getTrigger();
|
||||||
String status = SchedulerServiceConstants.STATUS_SUCCESS;
|
|
||||||
String errorMessage = null;
|
final JobKey key = context.getJobDetail().getKey();
|
||||||
String errorLog = null;
|
final String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
|
||||||
if (jobException != null) {
|
final ScheduledJobDetail scheduledJobDetails = this.schedularService.findByJobKey(jobKey);
|
||||||
status = SchedulerServiceConstants.STATUS_FAILED;
|
final Long version = this.schedularService.fetchMaxVersionBy(jobKey) + 1;
|
||||||
this.stackTraceLevel = 0;
|
String status = SchedulerServiceConstants.STATUS_SUCCESS;
|
||||||
final Throwable throwable = getCauseFromException(jobException);
|
String errorMessage = null;
|
||||||
this.stackTraceLevel = 0;
|
String errorLog = null;
|
||||||
StackTraceElement[] stackTraceElements = null;
|
if (jobException != null) {
|
||||||
errorMessage = throwable.getMessage();
|
status = SchedulerServiceConstants.STATUS_FAILED;
|
||||||
stackTraceElements = throwable.getStackTrace();
|
this.stackTraceLevel = 0;
|
||||||
final StringBuilder sb = new StringBuilder(throwable.toString());
|
final Throwable throwable = getCauseFromException(jobException);
|
||||||
for (final StackTraceElement element : stackTraceElements) {
|
this.stackTraceLevel = 0;
|
||||||
sb.append("\n \t at ").append(element.getClassName()).append(".").append(element.getMethodName()).append("(")
|
StackTraceElement[] stackTraceElements = null;
|
||||||
.append(element.getLineNumber()).append(")");
|
errorMessage = throwable.getMessage();
|
||||||
|
stackTraceElements = throwable.getStackTrace();
|
||||||
|
final StringBuilder sb = new StringBuilder(throwable.toString());
|
||||||
|
for (final StackTraceElement element : stackTraceElements) {
|
||||||
|
sb.append("\n \t at ").append(element.getClassName()).append(".").append(element.getMethodName()).append("(")
|
||||||
|
.append(element.getLineNumber()).append(")");
|
||||||
|
}
|
||||||
|
errorLog = sb.toString();
|
||||||
|
|
||||||
|
}
|
||||||
|
String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON;
|
||||||
|
if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) {
|
||||||
|
triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE);
|
||||||
|
}
|
||||||
|
if (SchedulerServiceConstants.TRIGGER_TYPE_CRON.equals(triggerType) && trigger.getNextFireTime() != null
|
||||||
|
&& trigger.getNextFireTime().after(scheduledJobDetails.getNextRunTime())) {
|
||||||
|
scheduledJobDetails.setNextRunTime(trigger.getNextFireTime());
|
||||||
}
|
}
|
||||||
errorLog = sb.toString();
|
|
||||||
|
|
||||||
|
scheduledJobDetails.setPreviousRunStartTime(context.getFireTime());
|
||||||
|
scheduledJobDetails.setCurrentlyRunning(false);
|
||||||
|
|
||||||
|
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();
|
||||||
}
|
}
|
||||||
String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON;
|
|
||||||
if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) {
|
|
||||||
triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE);
|
|
||||||
}
|
|
||||||
if (SchedulerServiceConstants.TRIGGER_TYPE_CRON.equals(triggerType) && trigger.getNextFireTime() != null
|
|
||||||
&& trigger.getNextFireTime().after(scheduledJobDetails.getNextRunTime())) {
|
|
||||||
scheduledJobDetails.setNextRunTime(trigger.getNextFireTime());
|
|
||||||
}
|
|
||||||
|
|
||||||
scheduledJobDetails.setPreviousRunStartTime(context.getFireTime());
|
|
||||||
scheduledJobDetails.setCurrentlyRunning(false);
|
|
||||||
|
|
||||||
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);
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private Throwable getCauseFromException(final Throwable exception) {
|
private Throwable getCauseFromException(final Throwable exception) {
|
||||||
|
|||||||
+7
-3
@@ -52,9 +52,13 @@ public class SchedulerTriggerListener implements TriggerListener {
|
|||||||
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);
|
String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
|
||||||
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
|
FineractPlatformTenant tenant = tenantDetailsService.loadTenantById(tenantIdentifier);
|
||||||
ThreadLocalContextUtil.setTenant(tenant);
|
try {
|
||||||
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
ThreadLocalContextUtil.setTenant(tenant);
|
||||||
return schedulerVetoer.veto(trigger, context);
|
ThreadLocalContextUtil.setActionContext(ActionContext.DEFAULT);
|
||||||
|
return schedulerVetoer.veto(trigger, context);
|
||||||
|
} finally {
|
||||||
|
ThreadLocalContextUtil.reset();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|||||||
+2
@@ -128,6 +128,8 @@ public class ClasspathDuplicatesStepDefinitions implements En {
|
|||||||
|| resourcePath.startsWith("META-INF/terracotta") || resourcePath.startsWith("com/fasterxml/jackson/core/io/doubleparser")
|
|| resourcePath.startsWith("META-INF/terracotta") || resourcePath.startsWith("com/fasterxml/jackson/core/io/doubleparser")
|
||||||
// Groovy is groovy
|
// Groovy is groovy
|
||||||
|| resourcePath.startsWith("META-INF/groovy")
|
|| resourcePath.startsWith("META-INF/groovy")
|
||||||
|
|| resourcePath.startsWith("org/springframework/batch/core/scope/context/JobSynchronizationManager")
|
||||||
|
|| resourcePath.startsWith("org/springframework/batch/core/scope/context/StepSynchronizationManager")
|
||||||
// Something doesn't to be a perfectly clean in Maven Surefire:
|
// Something doesn't to be a perfectly clean in Maven Surefire:
|
||||||
|| resourcePath.startsWith("META-INF/maven/") || resourcePath.contains("surefire")
|
|| resourcePath.startsWith("META-INF/maven/") || resourcePath.contains("surefire")
|
||||||
// org.slf4j.impl.StaticLoggerBinder.class in testutils for the
|
// org.slf4j.impl.StaticLoggerBinder.class in testutils for the
|
||||||
|
|||||||
+23
-2
@@ -19,14 +19,19 @@
|
|||||||
package org.apache.fineract.infrastructure.jobs.service;
|
package org.apache.fineract.infrastructure.jobs.service;
|
||||||
|
|
||||||
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
import static org.mockito.Mockito.times;
|
import static org.mockito.Mockito.times;
|
||||||
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.verify;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.stream.Stream;
|
import java.util.stream.Stream;
|
||||||
|
import org.apache.fineract.infrastructure.businessdate.service.BusinessDateReadPlatformService;
|
||||||
|
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
|
||||||
|
import org.apache.fineract.infrastructure.core.service.tenant.TenantDetailsService;
|
||||||
import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO;
|
import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO;
|
||||||
import org.apache.fineract.infrastructure.jobs.domain.JobParameter;
|
import org.apache.fineract.infrastructure.jobs.domain.JobParameter;
|
||||||
import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository;
|
import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository;
|
||||||
@@ -34,6 +39,8 @@ import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail;
|
|||||||
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameData;
|
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameData;
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService;
|
import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService;
|
||||||
import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider;
|
import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider;
|
||||||
|
import org.apache.fineract.useradministration.domain.AppUser;
|
||||||
|
import org.apache.fineract.useradministration.domain.AppUserRepositoryWrapper;
|
||||||
import org.junit.jupiter.api.Assertions;
|
import org.junit.jupiter.api.Assertions;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import org.junit.jupiter.api.extension.ExtendWith;
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
@@ -73,6 +80,14 @@ public class JobStarterTest {
|
|||||||
private List<JobParameterProvider<?>> jobParameterProviders;
|
private List<JobParameterProvider<?>> jobParameterProviders;
|
||||||
@Mock
|
@Mock
|
||||||
private JobNameService jobNameService;
|
private JobNameService jobNameService;
|
||||||
|
|
||||||
|
@Mock
|
||||||
|
private TenantDetailsService tenantDetailsService;
|
||||||
|
@Mock
|
||||||
|
private AppUserRepositoryWrapper userRepository;
|
||||||
|
@Mock
|
||||||
|
private BusinessDateReadPlatformService businessDateReadPlatformService;
|
||||||
|
|
||||||
@Captor
|
@Captor
|
||||||
private ArgumentCaptor<Set<JobParameterDTO>> jobParameterDTOCaptor;
|
private ArgumentCaptor<Set<JobParameterDTO>> jobParameterDTOCaptor;
|
||||||
|
|
||||||
@@ -106,7 +121,7 @@ public class JobStarterTest {
|
|||||||
ScheduledJobDetail scheduledJobDetail = Mockito.mock(ScheduledJobDetail.class);
|
ScheduledJobDetail scheduledJobDetail = Mockito.mock(ScheduledJobDetail.class);
|
||||||
when(jobExecution.getStatus()).thenReturn(BatchStatus.COMPLETED);
|
when(jobExecution.getStatus()).thenReturn(BatchStatus.COMPLETED);
|
||||||
setupMocks(jobExecution, job, scheduledJobDetail);
|
setupMocks(jobExecution, job, scheduledJobDetail);
|
||||||
JobExecution result = underTest.run(job, scheduledJobDetail, Set.of());
|
JobExecution result = underTest.run(job, scheduledJobDetail, Set.of(), "default");
|
||||||
Assertions.assertEquals(jobExecution, result);
|
Assertions.assertEquals(jobExecution, result);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -122,7 +137,7 @@ public class JobStarterTest {
|
|||||||
when(jobExecution.getStatus()).thenReturn(BatchStatus.FAILED);
|
when(jobExecution.getStatus()).thenReturn(BatchStatus.FAILED);
|
||||||
when(jobExecution.getExitStatus()).thenReturn(new ExitStatus(failedStatus.name(), "testException"));
|
when(jobExecution.getExitStatus()).thenReturn(new ExitStatus(failedStatus.name(), "testException"));
|
||||||
JobExecutionException exception = Assertions.assertThrows(JobExecutionException.class,
|
JobExecutionException exception = Assertions.assertThrows(JobExecutionException.class,
|
||||||
() -> underTest.run(job, scheduledJobDetail, Set.of()));
|
() -> underTest.run(job, scheduledJobDetail, Set.of(), "default"));
|
||||||
Assertions.assertEquals(String.format("exitCode=%s;exitDescription=%s", failedStatus.name(), "testException"),
|
Assertions.assertEquals(String.format("exitCode=%s;exitDescription=%s", failedStatus.name(), "testException"),
|
||||||
exception.getMessage());
|
exception.getMessage());
|
||||||
}
|
}
|
||||||
@@ -140,5 +155,11 @@ public class JobStarterTest {
|
|||||||
when(jobParameterProvider.canProvideParametersForJob("testJobName")).thenReturn(true);
|
when(jobParameterProvider.canProvideParametersForJob("testJobName")).thenReturn(true);
|
||||||
when(jobParameterProviders.stream()).thenReturn(Stream.of(jobParameterProvider));
|
when(jobParameterProviders.stream()).thenReturn(Stream.of(jobParameterProvider));
|
||||||
when(jobNameService.getJobByHumanReadableName(any(String.class))).thenReturn(new JobNameData("testEnumstyleName", "testHumanReadableName"));
|
when(jobNameService.getJobByHumanReadableName(any(String.class))).thenReturn(new JobNameData("testEnumstyleName", "testHumanReadableName"));
|
||||||
|
when(tenantDetailsService.loadTenantById(anyString())).thenReturn(FineractPlatformTenant.builder().build());
|
||||||
|
AppUser appUser = Mockito.mock(AppUser.class);
|
||||||
|
when(appUser.getPassword()).thenReturn("");
|
||||||
|
when(appUser.getAuthorities()).thenReturn(List.of());
|
||||||
|
when(userRepository.fetchSystemUser()).thenReturn(appUser);
|
||||||
|
when(businessDateReadPlatformService.getBusinessDates()).thenReturn(new HashMap<>());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user