SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Capture external-execution-id for child tasks started from a partition

Browse Source
This commit is contained in:
Glenn Renfro
2019-07-24 17:02:30 -04:00
committed by Michael Minella
parent 068df27723
commit 79719c541f
3 changed files with 111 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandler.java
View File

@@ -39,11 +39,13 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.poller.DirectPoller;
import org.springframework.batch.poller.Poller;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.deployer.spi.core.AppDefinition;
import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
import org.springframework.cloud.deployer.spi.task.TaskLauncher;
import org.springframework.cloud.task.listener.annotation.BeforeTask;
import org.springframework.cloud.task.repository.TaskExecution;
import org.springframework.cloud.task.repository.TaskRepository;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.core.io.Resource;
@@ -72,6 +74,7 @@ import org.springframework.util.StringUtils;
* </p>
*
* @author Michael Minella
* @author Glenn Renfro
*/
public class DeployerPartitionHandler
implements PartitionHandler, EnvironmentAware, InitializingBean {
@@ -96,6 +99,11 @@ public class DeployerPartitionHandler
*/
public static final String SPRING_CLOUD_TASK_PARENT_EXECUTION_ID = "spring.cloud.task.parentExecutionId";
/**
* ID of the Spring Cloud Task execution.
*/
public static final String SPRING_CLOUD_TASK_EXECUTION_ID = "spring.cloud.task.executionid";
/**
* Spring Cloud Task name property.
*/
@@ -137,6 +145,19 @@ public class DeployerPartitionHandler
private boolean defaultArgsAsEnvironmentVars = false;
@Autowired
private TaskRepository taskRepository;
/**
* Constructor initializing the DeployerPartitionHandler instance.
* @param taskLauncher the launcher used to execute partitioned tasks.
* @param jobExplorer used to acquire the status of the job.
* @param resource the url to the app to be launched.
* @param stepName the name of the step.
* @deprecated Use the constructor that accepts {@link TaskRepository} as well as the
* this constructor's set of parameters.
*/
@Deprecated
public DeployerPartitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer,
Resource resource, String stepName) {
Assert.notNull(taskLauncher, "A taskLauncher is required");
@@ -150,6 +171,21 @@ public class DeployerPartitionHandler
this.stepName = stepName;
}
public DeployerPartitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer,
Resource resource, String stepName, TaskRepository taskRepository) {
Assert.notNull(taskLauncher, "A taskLauncher is required");
Assert.notNull(jobExplorer, "A jobExplorer is required");
Assert.notNull(resource, "A resource is required");
Assert.hasText(stepName, "A step name is required");
Assert.notNull(taskRepository, "A TaskRepository is required");
this.taskLauncher = taskLauncher;
this.jobExplorer = jobExplorer;
this.resource = resource;
this.stepName = stepName;
this.taskRepository = taskRepository;
}
/**
* Used to provide any environment variables to be set on each worker launched.
* @param environmentVariablesProvider an {@link EnvironmentVariablesProvider}
@@ -290,6 +326,8 @@ public class DeployerPartitionHandler
arguments.addAll(this.commandLineArgsProvider.getCommandLineArgs(copyContext));
TaskExecution partitionTaskExecution = this.taskRepository.createTaskExecution();
if (!this.defaultArgsAsEnvironmentVars) {
arguments.add(formatArgument(SPRING_CLOUD_TASK_JOB_EXECUTION_ID,
String.valueOf(workerStepExecution.getJobExecution().getId())));
@@ -304,6 +342,8 @@ public class DeployerPartitionHandler
workerStepExecution.getStepName())));
arguments.add(formatArgument(SPRING_CLOUD_TASK_PARENT_EXECUTION_ID,
String.valueOf(this.taskExecution.getExecutionId())));
arguments.add(formatArgument(SPRING_CLOUD_TASK_EXECUTION_ID,
String.valueOf(partitionTaskExecution.getExecutionId())));
}
copyContext = new ExecutionContext(workerStepExecution.getExecutionContext());
@@ -325,6 +365,8 @@ public class DeployerPartitionHandler
workerStepExecution.getStepName()));
environmentVariables.put(SPRING_CLOUD_TASK_PARENT_EXECUTION_ID,
String.valueOf(this.taskExecution.getExecutionId()));
environmentVariables.put(SPRING_CLOUD_TASK_EXECUTION_ID,
String.valueOf(partitionTaskExecution.getExecutionId()));
}
AppDefinition definition = new AppDefinition(resolveApplicationName(),
@@ -333,7 +375,9 @@ public class DeployerPartitionHandler
AppDeploymentRequest request = new AppDeploymentRequest(definition, this.resource,
this.deploymentProperties, arguments);
this.taskLauncher.launch(request);
String externalExecutionId = this.taskLauncher.launch(request);
this.taskRepository.updateExternalExecutionId(
partitionTaskExecution.getExecutionId(), externalExecutionId);
}
private String resolveApplicationName() {

87
spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandlerTests.java
View File

@@ -42,11 +42,11 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.cloud.deployer.spi.core.AppDefinition;
import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
import org.springframework.cloud.deployer.spi.task.TaskLauncher;
import org.springframework.cloud.task.repository.TaskExecution;
import org.springframework.cloud.task.repository.TaskRepository;
import org.springframework.core.env.Environment;
import org.springframework.core.io.Resource;
import org.springframework.mock.env.MockEnvironment;
@@ -60,6 +60,7 @@ import static org.mockito.Mockito.when;
/**
* @author Michael Minella
* @author Glenn Renfro
*/
public class DeployerPartitionHandlerTests {
@@ -79,7 +80,7 @@ public class DeployerPartitionHandlerTests {
private StepExecutionSplitter splitter;
@Mock
private JobRepository jobRepository;
private TaskRepository taskRepository;
private Environment environment;
@@ -87,21 +88,42 @@ public class DeployerPartitionHandlerTests {
public void setUp() {
MockitoAnnotations.initMocks(this);
this.environment = new MockEnvironment();
TaskExecution taskExecution = new TaskExecution(2, 0, "name", new Date(),
new Date(), "", Collections.emptyList(), null, null, null);
when(taskRepository.createTaskExecution()).thenReturn(taskExecution);
}
@Test
public void testDeprecatedConstructorValidation() {
validateDeprecatedConstructorValidation(null, null, null, null,
"A taskLauncher is required");
validateDeprecatedConstructorValidation(this.taskLauncher, null, null, null,
"A jobExplorer is required");
validateDeprecatedConstructorValidation(this.taskLauncher, this.jobExplorer, null,
null, "A resource is required");
validateDeprecatedConstructorValidation(this.taskLauncher, this.jobExplorer,
this.resource, null, "A step name is required");
new DeployerPartitionHandler(this.taskLauncher, this.jobExplorer, this.resource,
"step-name");
}
@Test
public void testConstructorValidation() {
validateConstructorValidation(null, null, null, null,
validateConstructorValidation(null, null, null, null, null,
"A taskLauncher is required");
validateConstructorValidation(this.taskLauncher, null, null, null,
validateConstructorValidation(this.taskLauncher, null, null, null, null,
"A jobExplorer is required");
validateConstructorValidation(this.taskLauncher, this.jobExplorer, null, null,
"A resource is required");
null, "A resource is required");
validateConstructorValidation(this.taskLauncher, this.jobExplorer, this.resource,
null, "A step name is required");
null, null, "A step name is required");
validateConstructorValidation(this.taskLauncher, this.jobExplorer, this.resource,
null, null, "A step name is required");
validateConstructorValidation(this.taskLauncher, this.jobExplorer, this.resource,
"step-name", null, "A TaskRepository is required");
new DeployerPartitionHandler(this.taskLauncher, this.jobExplorer, this.resource,
"step-name");
"step-name", this.taskRepository);
}
@Test
@@ -131,7 +153,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
TaskExecution taskExecution = new TaskExecution();
@@ -170,6 +192,8 @@ public class DeployerPartitionHandlerTests {
.isTrue();
assertThat(request.getCommandlineArguments().contains(formatArgs(
DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1"))).isTrue();
assertThat(request.getCommandlineArguments()
.contains(formatArgs("spring.cloud.task.executionid", "2"))).isTrue();
assertThat(results.size()).isEqualTo(1);
StepExecution resultStepExecution = results.iterator().next();
@@ -186,9 +210,8 @@ public class DeployerPartitionHandlerTests {
StepExecution workerStepExecutionStart = getStepExecutionStart(jobExecution, 4L);
StepExecution workerStepExecutionFinish = getStepExecutionFinish(
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setDefaultArgsAsEnvironmentVars(true);
@@ -237,6 +260,9 @@ public class DeployerPartitionHandlerTests {
assertThat(request.getDefinition().getProperties()
.get(DeployerPartitionHandler.SPRING_CLOUD_TASK_PARENT_EXECUTION_ID))
.isEqualTo("55");
assertThat(request.getDefinition().getProperties()
.get(DeployerPartitionHandler.SPRING_CLOUD_TASK_EXECUTION_ID))
.isEqualTo("2");
assertThat(results.size()).isEqualTo(1);
StepExecution resultStepExecution = results.iterator().next();
@@ -255,7 +281,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
TaskExecution taskExecution = new TaskExecution(55, null, null, null, null, null,
@@ -304,7 +330,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart3, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
TaskExecution taskExecution = new TaskExecution();
@@ -361,7 +387,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart3, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setMaxWorkers(2);
@@ -419,7 +445,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart3, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setMaxWorkers(2);
@@ -484,7 +510,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
Map<String, String> environmentParameters = new HashMap<>(2);
environmentParameters.put("foo", "bar");
@@ -553,7 +579,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
Map<String, String> environmentParameters = new HashMap<>(2);
@@ -626,7 +652,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart2, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setPollInterval(20000L);
@@ -684,7 +710,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart2, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setPollInterval(20000L);
@@ -726,7 +752,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart2, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
handler.setGridSize(2);
@@ -773,7 +799,7 @@ public class DeployerPartitionHandlerTests {
workerStepExecutionStart, BatchStatus.COMPLETED);
DeployerPartitionHandler handler = new DeployerPartitionHandler(this.taskLauncher,
this.jobExplorer, this.resource, "step1");
this.jobExplorer, this.resource, "step1", this.taskRepository);
handler.setEnvironment(this.environment);
Map<String, String> deploymentProperties = new HashMap<>(2);
@@ -919,14 +945,29 @@ public class DeployerPartitionHandlerTests {
assertThat(request.getCommandlineArguments().contains(formatArgs(
DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")))
.isTrue();
assertThat(request.getCommandlineArguments()
.contains(formatArgs("spring.cloud.task.executionid", "2"))).isTrue();
}
}
private void validateDeprecatedConstructorValidation(TaskLauncher taskLauncher,
JobExplorer jobExplorer, Resource resource, String stepName,
String expectedMessage) {
try {
new DeployerPartitionHandler(taskLauncher, jobExplorer, resource, stepName,
this.taskRepository);
}
catch (IllegalArgumentException iae) {
assertThat(iae.getMessage()).isEqualTo(expectedMessage);
}
}
private void validateConstructorValidation(TaskLauncher taskLauncher,
JobExplorer jobExplorer, Resource resource, String stepName,
String expectedMessage) {
TaskRepository taskRepository, String expectedMessage) {
try {
new DeployerPartitionHandler(taskLauncher, jobExplorer, resource, stepName);
new DeployerPartitionHandler(taskLauncher, jobExplorer, resource, stepName,
taskRepository);
}
catch (IllegalArgumentException iae) {
assertThat(iae.getMessage()).isEqualTo(expectedMessage);