Fix unfinished step in parallel flow

Resolves #3939

Signed-off-by: Mahmoud Ben Hassine <mbenhassine@vmware.com>
This commit is contained in:
doontagi
2024-03-24 22:55:09 +09:00
committed by Mahmoud Ben Hassine
parent bd00cde09f
commit aafecb93d4
2 changed files with 49 additions and 5 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2023 the original author or authors.
* Copyright 2006-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
@@ -119,7 +120,7 @@ public class SplitState extends AbstractState implements FlowHolder {
FlowExecutionStatus parentSplitStatus = parentSplit == null ? null : parentSplit.handle(executor);
Collection<FlowExecution> results = new ArrayList<>();
List<Exception> exceptions = new ArrayList<>();
// Could use a CompletionService here?
for (Future<FlowExecution> task : tasks) {
try {
@@ -129,14 +130,18 @@ public class SplitState extends AbstractState implements FlowHolder {
// Unwrap the expected exceptions
Throwable cause = e.getCause();
if (cause instanceof Exception) {
throw (Exception) cause;
exceptions.add((Exception) cause);
}
else {
throw e;
exceptions.add(e);
}
}
}
if (!exceptions.isEmpty()) {
throw exceptions.get(0);
}
FlowExecutionStatus flowExecutionStatus = doAggregation(results, executor);
if (parentSplitStatus != null) {
return Collections.max(Arrays.asList(flowExecutionStatus, parentSplitStatus));

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2023 the original author or authors.
* Copyright 2012-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,14 @@
package org.springframework.batch.core.job.builder;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.sql.DataSource;
import static org.junit.jupiter.api.Assertions.assertEquals;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.batch.core.BatchStatus;
@@ -45,6 +48,8 @@ import org.springframework.batch.core.repository.support.JobRepositoryFactoryBea
import org.springframework.batch.core.step.StepSupport;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -369,4 +374,38 @@ class FlowJobBuilderTests {
}
@Test
public void testBuildSplitWithParallelFlow() throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(1);
Step longExecutingStep = new StepBuilder("longExecutingStep", jobRepository).tasklet((stepContribution, b) -> {
Thread.sleep(500L);
return RepeatStatus.FINISHED;
}, new ResourcelessTransactionManager()).build();
Step interruptedStep = new StepBuilder("interruptedStep", jobRepository).tasklet((stepContribution, b) -> {
stepContribution.getStepExecution().setTerminateOnly();
return RepeatStatus.FINISHED;
}, new ResourcelessTransactionManager()).build();
Step nonExecutableStep = new StepBuilder("nonExecutableStep", jobRepository).tasklet((stepContribution, b) -> {
countDownLatch.countDown();
return RepeatStatus.FINISHED;
}, new ResourcelessTransactionManager()).build();
Flow twoStepFlow = new FlowBuilder<SimpleFlow>("twoStepFlow").start(longExecutingStep)
.next(nonExecutableStep)
.build();
Flow interruptedFlow = new FlowBuilder<SimpleFlow>("interruptedFlow").start(interruptedStep).build();
Flow splitFlow = new FlowBuilder<Flow>("splitFlow").split(new SimpleAsyncTaskExecutor())
.add(interruptedFlow, twoStepFlow)
.build();
FlowJobBuilder jobBuilder = new JobBuilder("job", jobRepository).start(splitFlow).build();
jobBuilder.preventRestart().build().execute(execution);
boolean isExecutedNonExecutableStep = countDownLatch.await(1, TimeUnit.SECONDS);
assertEquals(BatchStatus.STOPPED, execution.getStatus());
Assertions.assertFalse(isExecutedNonExecutableStep);
}
}