diff --git a/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java b/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java index f8376f9b9..ece705203 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java +++ b/fineract-provider/src/main/java/org/apache/fineract/cob/service/AsyncLoanCOBExecutorServiceImpl.java @@ -25,6 +25,7 @@ import java.util.HashSet; import java.util.List; import java.util.Set; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.apache.fineract.cob.conditions.LoanCOBEnabledCondition; import org.apache.fineract.cob.data.LoanIdAndLastClosedBusinessDate; import org.apache.fineract.cob.loan.LoanCOBConstant; @@ -40,6 +41,7 @@ import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetailRepository; import org.apache.fineract.infrastructure.jobs.exception.JobNotFoundException; import org.apache.fineract.infrastructure.jobs.service.JobStarter; +import org.quartz.JobExecutionException; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobParametersInvalidException; import org.springframework.batch.core.configuration.JobLocator; @@ -51,6 +53,7 @@ import org.springframework.context.annotation.Conditional; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; +@Slf4j @Service @RequiredArgsConstructor @Conditional(LoanCOBEnabledCondition.class) @@ -78,10 +81,12 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ executeLoanCOBDayByDayUntilCOBBusinessDate(oldestCOBProcessedDate, cobBusinessDate); } } catch (NoSuchJobException e) { - throw new JobNotFoundException(LoanCOBConstant.JOB_NAME, e); + // Throwing an error here is useless as it will be swallowed hence it is async method + log.error("", new JobNotFoundException(LoanCOBConstant.JOB_NAME, e)); } catch (JobInstanceAlreadyCompleteException | JobRestartException | JobParametersInvalidException - | JobExecutionAlreadyRunningException e) { - throw new RuntimeException(e); + | JobExecutionAlreadyRunningException | JobExecutionException e) { + // Throwing an error here is useless as it will be swallowed hence it is async method + log.error("", e); } finally { ThreadLocalContextUtil.reset(); } @@ -89,7 +94,7 @@ public class AsyncLoanCOBExecutorServiceImpl implements AsyncLoanCOBExecutorServ private void executeLoanCOBDayByDayUntilCOBBusinessDate(LocalDate oldestCOBProcessedDate, LocalDate cobBusinessDate) throws NoSuchJobException, JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, - JobParametersInvalidException, JobRestartException { + JobParametersInvalidException, JobRestartException, JobExecutionException { Job job = jobLocator.getJob(LoanCOBConstant.JOB_NAME); ScheduledJobDetail scheduledJobDetail = scheduledJobDetailRepository.findByJobName(LoanCOBConstant.JOB_HUMAN_READABLE_NAME); LocalDate executingBusinessDate = oldestCOBProcessedDate.plusDays(1); diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java index 02674ac4f..031dc0737 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/JobStarter.java @@ -32,7 +32,10 @@ import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository; import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService; import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider; +import org.quartz.JobExecutionException; +import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameter; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersBuilder; @@ -55,19 +58,26 @@ public class JobStarter { private final List> jobParameterProviders; private final JobNameService jobNameService; - public void run(Job job, ScheduledJobDetail scheduledJobDetail, Set jobParameterDTOSet) + public static final List FAILED_STATUSES = List.of(BatchStatus.FAILED, BatchStatus.ABANDONED, BatchStatus.STOPPED, + BatchStatus.STOPPING, BatchStatus.UNKNOWN); + + public JobExecution run(Job job, ScheduledJobDetail scheduledJobDetail, Set jobParameterDTOSet) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, - JobRestartException { + JobRestartException, JobExecutionException { Map> 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(); - jobLauncher.run(job, jobParameters); + JobExecution result = jobLauncher.run(job, jobParameters); + if (FAILED_STATUSES.contains(result.getStatus())) { + throw new JobExecutionException(result.getExitStatus().toString()); + } + return result; } - public Map> getJobParameter(ScheduledJobDetail scheduledJobDetail) { + protected Map> getJobParameter(ScheduledJobDetail scheduledJobDetail) { List jobParameterList = jobParameterRepository .findJobParametersByJobId(scheduledJobDetail.getId()); Map> jobParameterMap = new HashMap<>(); @@ -77,7 +87,7 @@ public class JobStarter { return jobParameterMap; } - private Map> provideCustomJobParameters(String jobName, Set jobParameterDTOSet) { + protected Map> provideCustomJobParameters(String jobName, Set jobParameterDTOSet) { Optional> jobParameterProvider = jobParameterProviders.stream() .filter(provider -> provider.canProvideParametersForJob(jobName)).findFirst(); Map> map = jobParameterProvider diff --git a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/jobs/service/JobStarterTest.java b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/jobs/service/JobStarterTest.java new file mode 100644 index 000000000..a9bec4171 --- /dev/null +++ b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/jobs/service/JobStarterTest.java @@ -0,0 +1,144 @@ +/** + * 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.service; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; +import org.apache.fineract.infrastructure.jobs.data.JobParameterDTO; +import org.apache.fineract.infrastructure.jobs.domain.JobParameter; +import org.apache.fineract.infrastructure.jobs.domain.JobParameterRepository; +import org.apache.fineract.infrastructure.jobs.domain.ScheduledJobDetail; +import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameData; +import org.apache.fineract.infrastructure.jobs.service.jobname.JobNameService; +import org.apache.fineract.infrastructure.jobs.service.jobparameterprovider.JobParameterProvider; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; +import org.quartz.JobExecutionException; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersIncrementer; +import org.springframework.batch.core.JobParametersInvalidException; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRestartException; + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +public class JobStarterTest { + + @Mock + private JobExplorer jobExplorer; + @Mock + private JobLauncher jobLauncher; + @Mock + private JobParameterRepository jobParameterRepository; + @Mock + private List> jobParameterProviders; + @Mock + private JobNameService jobNameService; + @Captor + private ArgumentCaptor> jobParameterDTOCaptor; + + @InjectMocks + private JobStarter underTest; + + @Test + public void getJobParameterTest() { + ScheduledJobDetail scheduledJobDetail = Mockito.mock(ScheduledJobDetail.class); + when(scheduledJobDetail.getId()).thenReturn(1L); + when(jobParameterRepository.findJobParametersByJobId(1L)) + .thenReturn(List.of(new JobParameter().setJobId(1L).setParameterName("testParamKey").setParameterValue("testParamValue"))); + Map> result = underTest.getJobParameter(scheduledJobDetail); + Assertions.assertEquals("testParamValue", result.get("testParamKey").getValue()); + } + + @Test + public void provideCustomJobParameters() { + JobParameterProvider jobParameterProvider = Mockito.mock(JobParameterProvider.class); + when(jobParameterProvider.canProvideParametersForJob("testJobName")).thenReturn(true); + when(jobParameterProviders.stream()).thenReturn(Stream.of(jobParameterProvider)); + underTest.provideCustomJobParameters("testJobName", Set.of(new JobParameterDTO("testKey", "testValue"))); + verify(jobParameterProvider, times(1)).provide(jobParameterDTOCaptor.capture()); + } + + @Test + public void runWithComplete() throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, + JobParametersInvalidException, JobRestartException, JobExecutionException { + JobExecution jobExecution = Mockito.mock(JobExecution.class); + Job job = Mockito.mock(Job.class); + ScheduledJobDetail scheduledJobDetail = Mockito.mock(ScheduledJobDetail.class); + when(jobExecution.getStatus()).thenReturn(BatchStatus.COMPLETED); + setupMocks(jobExecution, job, scheduledJobDetail); + JobExecution result = underTest.run(job, scheduledJobDetail, Set.of()); + Assertions.assertEquals(jobExecution, result); + } + + @Test + public void runWithFailed() throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, + JobParametersInvalidException, JobRestartException, JobExecutionException { + JobExecution jobExecution = Mockito.mock(JobExecution.class); + Job job = Mockito.mock(Job.class); + ScheduledJobDetail scheduledJobDetail = Mockito.mock(ScheduledJobDetail.class); + + for (BatchStatus failedStatus : JobStarter.FAILED_STATUSES) { + setupMocks(jobExecution, job, scheduledJobDetail); + when(jobExecution.getStatus()).thenReturn(BatchStatus.FAILED); + when(jobExecution.getExitStatus()).thenReturn(new ExitStatus(failedStatus.name(), "testException")); + JobExecutionException exception = Assertions.assertThrows(JobExecutionException.class, + () -> underTest.run(job, scheduledJobDetail, Set.of())); + Assertions.assertEquals(String.format("exitCode=%s;exitDescription=%s", failedStatus.name(), "testException"), + exception.getMessage()); + } + } + + private void setupMocks(JobExecution jobExecution, Job job, ScheduledJobDetail scheduledJobDetail) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException { + when(scheduledJobDetail.getId()).thenReturn(1L); + when(scheduledJobDetail.getJobName()).thenReturn("testJobName"); + when(jobParameterRepository.findJobParametersByJobId(1L)).thenReturn(List.of(new JobParameter().setJobId(1L).setParameterName("testParamKey").setParameterValue("testParamValue"))); + when(jobLauncher.run(any(Job.class), any(JobParameters.class))).thenReturn(jobExecution); + JobParametersIncrementer jobParametersIncrementer = Mockito.mock(JobParametersIncrementer.class); + when(jobParametersIncrementer.getNext(any(JobParameters.class))).thenReturn(new JobParameters()); + when(job.getJobParametersIncrementer()).thenReturn(jobParametersIncrementer); + JobParameterProvider jobParameterProvider = Mockito.mock(JobParameterProvider.class); + when(jobParameterProvider.canProvideParametersForJob("testJobName")).thenReturn(true); + when(jobParameterProviders.stream()).thenReturn(Stream.of(jobParameterProvider)); + when(jobNameService.getJobByHumanReadableName(any(String.class))).thenReturn(new JobNameData("testEnumstyleName", "testHumanReadableName")); + } +}