From 79719c541f4619483490d9e9e95fc48fce8d30a7 Mon Sep 17 00:00:00 2001 From: Glenn Renfro Date: Wed, 24 Jul 2019 17:02:30 -0400 Subject: [PATCH] Capture external-execution-id for child tasks started from a partition --- .../partition/DeployerPartitionHandler.java | 46 +++++++++- .../DeployerPartitionHandlerTests.java | 87 ++++++++++++++----- .../main/java/io/spring/JobConfiguration.java | 3 +- 3 files changed, 111 insertions(+), 25 deletions(-) diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandler.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandler.java index d4c091dc..fe225c82 100644 --- a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandler.java +++ b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandler.java @@ -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; *

* * @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() { diff --git a/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandlerTests.java b/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandlerTests.java index dac40b71..a0716696 100644 --- a/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandlerTests.java +++ b/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/DeployerPartitionHandlerTests.java @@ -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 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 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 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); diff --git a/spring-cloud-task-samples/partitioned-batch-job/src/main/java/io/spring/JobConfiguration.java b/spring-cloud-task-samples/partitioned-batch-job/src/main/java/io/spring/JobConfiguration.java index cd69b883..d0da0d3d 100644 --- a/spring-cloud-task-samples/partitioned-batch-job/src/main/java/io/spring/JobConfiguration.java +++ b/spring-cloud-task-samples/partitioned-batch-job/src/main/java/io/spring/JobConfiguration.java @@ -46,6 +46,7 @@ import org.springframework.cloud.task.batch.partition.DeployerPartitionHandler; import org.springframework.cloud.task.batch.partition.DeployerStepExecutionHandler; import org.springframework.cloud.task.batch.partition.PassThroughCommandLineArgsProvider; import org.springframework.cloud.task.batch.partition.SimpleEnvironmentVariablesProvider; +import org.springframework.cloud.task.repository.TaskRepository; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -78,7 +79,7 @@ public class JobConfiguration { private Environment environment; @Bean - public PartitionHandler partitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer) throws Exception { + public PartitionHandler partitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer, TaskRepository taskRepository) throws Exception { Resource resource = this.resourceLoader .getResource("maven://io.spring.cloud:partitioned-batch-job:2.2.0.BUILD-SNAPSHOT");