Spring批处理的集成测试因线程执行而失败

Question

我有一个Spring批处理项目。目标是对Spring作业中的各个步骤执行集成测试。

我正在使用JobLauncherTestUtils来启动步骤。但是，当这个util启动步骤时，它会在一个单独的线程中运行它。一旦线程完成执行，应该会有一些值分配给jobExecution.getStepExections（）。

问题：由于某些原因，即使在线程完成执行之前，测试也会转到下一行List

问题：有没有什么优雅的方法可以等待步骤执行线程完成，然后继续测试中的下一行进行验证？

代码：

集成测试类：

@Slf4j
@SpringBatchTest
@SpringBootTest
@ActiveProfiles({"test", "master"})
@ContextConfiguration(classes = {InhouseClass3.class, InhouseClass1.class, InhouseClass2.class})
public class BatchJobIntegrationTest {

    private static final String Param1 = "someParam";

    @Autowired
    @Qualifier("hikariDatasource")
    DataSource hikariDatasource;

    @Autowired
    Job BatchJob;

    @Autowired
    JobLauncher jobLauncher;

    JobExecution jobExecution;

    @Autowired
    CreateDirectoryTasklet createDirectoryTasklet;

    JobParameters jobParameters;

    JobLauncherTestUtils jobLauncherTestUtils;

    @BeforeEach
    void setUp() {
        String startTimestamp = Timestamp.from(Instant.now()).toString();
        jobParameters = new JobParametersBuilder()
                .addString(Param1, startTimestamp)
                .toJobParameters();

        jobLauncherTestUtils = new JobLauncherTestUtils();
        jobLauncherTestUtils.setJob(BatchJob);
        jobLauncherTestUtils.setJobLauncher(jobLauncher);
    }


    @SneakyThrows
    @Test
    void TaskletToTest_Test() {
        jobExecution = jobLauncherTestUtils.launchStep("loadTaskletToTest", jobParameters);

        List<StepExecution> actualStepExecutions = new ArrayList<>(jobExecution.getStepExecutions());
        // ERROR: jobExecution.getStepExecutions() is NULL. 

        ExitStatus actualJobExitStatus = actualStepExecutions.get(0).getExitStatus();
        // ERROR: Index 0 out of bounds for length 0

        assertEquals("loadGdxClaims", actualStepExecutions.get(0).getStepName());
        assertEquals(ExitStatus.COMPLETED, actualJobExitStatus);
    }

    @SneakyThrows
    @Test
    // This is my workaround to make my above test run. 
    // I added a sleep for 2 seconds. But this doesn't look like an ideal way, coz what 
    // if the launchstep thread running the tasklet took more than 2 seconds? 
    void loadGDXClaimTaskletTest_Working_() {
        jobExecution = jobLauncherTestUtils.launchStep("createDirectory", jobParameters);
        boolean counter = true;
        while(counter) {
            if (jobExecution.getStepExecutions().size()!=0 ) {
                List<StepExecution> actualStepExecutions = new ArrayList<>(jobExecution.getStepExecutions());

                ExitStatus actualJobExitStatus = actualStepExecutions.get(0).getExitStatus();
                log.info("------- step executions : {}", actualStepExecutions);
                assertEquals("createDirectory", actualStepExecutions.get(0).getStepName());
                assertEquals(ExitStatus.COMPLETED, actualJobExitStatus);
                counter = false;
            } else {
                TimeUnit.SECONDS.sleep(2);
            }
        }
    }



}

要测试的Tasklet步骤：

@Slf4j
@Component
public class TaskletToTest implements Tasklet {

    private final InhouseService inhouseService;

    public TaskletToTest(InhouseService inhouseService) {
        this.inhouseService = inhouseService;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws InterruptedException, IllegalJobNameException, JSchException, IOException {
        log.info("TaskletToTest before");
        inhouseService.retry();
        log.info("TaskletToTest after");
        return RepeatStatus.FINISHED;
    }
}

批处理作业，包括需要测试的步骤列表：

@Slf4j
@Profile("master")
@Configuration
public class MasterConfig extends DefaultBatchConfigurer {
    @Bean(name = "BatchJob")
    public Job remoteChunkingJob(TaskletStep someOtherTasklet,
                                 TaskletStep loadTaskletToTest,
                                 JobExecutionListener jobExecutionListener) {
        return this.jobBuilderFactory.get("extract gdx load")
                .incrementer(new RunIdIncrementer())     
                .listener(jobExecutionListener)
                .start(someOtherTasklet) 
                .next(loadTaskletToTest)
                .build();
    }

    @Bean
    TaskletStep loadTaskletToTest(TaskletToTest taskletToTest) {
        return this.stepBuilderFactory.get("loadTaskletToTest").tasklet(taskletToTest).build();
    }
}




Answer 1

我正在使用JobLauncherTestUtils启动步骤。然而，当这个util启动这些步骤时，它会在一个单独的线程中运行它。

问题：由于某种原因，即使在线程完成执行之前，测试也会转到下一行

JobLauncherTestUtils使用JobLauncher启动作业和步骤。因此，根据您使用的JobLauncher实现，您可以在当前线程或单独的线程中运行作业/步骤。

您没有分享哪个JobLauncher在您的测试中是自动加载的，但是您似乎已经定义了一个基于异步TaskExecator实现的作业启动器。这就是为什么您的作业/步骤在后台执行：

// This returns immediately with an asynchrnous TaskExecutor
// However, with a synchronous TaskExectuor it will block waiting for the step to finish 
jobExecution = jobLauncherTestUtils.launchStep("loadTaskletToTest", jobParameters);

List<StepExecution> actualStepExecutions = new ArrayList<>(jobExecution.getStepExecutions());
        

因此，您需要检查测试中自动连接的是哪个JobLauncher（通常是带有同步或异步TaskExecutor的SimpleJobLauncher）。