Refactored for CloudFoundry Support

This commit adds on an abstraction for the providing unique command line
arugements for each worker.  It also updates the partitioned job sample
to be able to successfully be run on CloudFoundry.

Resolves spring-cloud/spring-cloud-task#193

Updated based on comments

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/CommandLineArgsProvider.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.List;
+
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * Strategy to allow for the customization of command line arguments passed to each
+ * partition's execution.
+ *
+ * @author Michael Minella
+ * @since 1.1.0
+ */
+public interface CommandLineArgsProvider {
+
+	/**
+	 * Returns a unique list of command line arguements to be passed to the partition's
+	 * worker for the specified {@link ExecutionContext}.
+	 *
+	 * Note: This method is called once per partition.
+	 *
+	 * @param executionContext the unique state for the step to be executed.
+	 * @return a list of formatted command line arguments to be passed to the worker (the
+	 *         list will be joined via spaces).
+	 */
+	List<String> getCommandLineArgs(ExecutionContext executionContext);
+}

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

@@ -47,6 +47,7 @@ import org.springframework.core.env.Environment;
 import org.springframework.core.io.Resource;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * <p>A {@link PartitionHandler} implementation that delegates to a {@link TaskLauncher} for
@@ -75,6 +76,8 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 	public static final String SPRING_CLOUD_TASK_STEP_NAME =
 			"spring.cloud.task.step-name";
 
+	public static final String SPRING_CLOUD_TASK_NAME = "spring.cloud.task.name";
+
 	private int maxWorkers = -1;
 
 	private int gridSize = 1;
@@ -103,6 +106,10 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 
 	private EnvironmentVariablesProvider environmentVariablesProvider;
 
+	private String applicationName;
+
+	private CommandLineArgsProvider commandLineArgsProvider;
+
 	public DeployerPartitionHandler(TaskLauncher taskLauncher,
 			JobExplorer jobExplorer,
 			Resource resource,
@@ -118,10 +125,24 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 		this.stepName = stepName;
 	}
 
+	/**
+	 * Used to provide any environment variables to be set on each worker launched.
+	 *
+	 * @param environmentVariablesProvider an {@link EnvironmentVariablesProvider}
+	 */
 	public void setEnvironmentVariablesProvider(EnvironmentVariablesProvider environmentVariablesProvider) {
 		this.environmentVariablesProvider = environmentVariablesProvider;
 	}
 
+	/**
+	 * Used to provide any command line arguements to be passed to each worker launched.
+	 *
+	 * @param commandLineArgsProvider {@link CommandLineArgsProvider}
+	 */
+	public void setCommandLineArgsProvider(CommandLineArgsProvider commandLineArgsProvider) {
+		this.commandLineArgsProvider = commandLineArgsProvider;
+	}
+
 	/**
 	 * The maximum number of workers to be executing at once.
 	 *
@@ -170,9 +191,24 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 		this.deploymentProperties = deploymentProperties;
 	}
 
+	/**
+	 * The name of the application to be launched.  Useful in environments where
+	 * application deployments are reused (such as CloudFoundry).
+	 *
+	 * @param applicationName The name of the application to be launched
+	 */
+	public void setApplicationName(String applicationName) {
+		this.applicationName = applicationName;
+	}
+
 	@BeforeTask
 	public void beforeTask(TaskExecution taskExecution) {
 		this.taskExecution = taskExecution;
+
+		if(this.commandLineArgsProvider == null) {
+			this.commandLineArgsProvider =
+					new SimpleCommandLineArgsProvider(this.taskExecution);
+		}
 	}
 
 	@Override
@@ -216,21 +252,30 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 
 	private void launchWorker(StepExecution workerStepExecution) {
 		List<String> arguments = new ArrayList<>();
-		arguments.addAll(this.taskExecution.getArguments());
+
+		ExecutionContext copyContext = new ExecutionContext(workerStepExecution.getExecutionContext());
+
+		arguments.addAll(
+				this.commandLineArgsProvider
+						.getCommandLineArgs(copyContext));
+
 		arguments.add(formatArgument(SPRING_CLOUD_TASK_JOB_EXECUTION_ID,
 				String.valueOf(workerStepExecution.getJobExecution().getId())));
 		arguments.add(formatArgument(SPRING_CLOUD_TASK_STEP_EXECUTION_ID,
 				String.valueOf(workerStepExecution.getId())));
 		arguments.add(formatArgument(SPRING_CLOUD_TASK_STEP_NAME, this.stepName));
+		arguments.add(formatArgument(SPRING_CLOUD_TASK_NAME, String.format("%s:%s:%s",
+				taskExecution.getTaskName(),
+				workerStepExecution.getJobExecution().getJobInstance().getJobName(),
+				workerStepExecution.getStepName())));
 
-		ExecutionContext copyContext = new ExecutionContext(workerStepExecution.getExecutionContext());
+		copyContext = new ExecutionContext(workerStepExecution.getExecutionContext());
+
+		Map<String, String> environmentVariables = this.environmentVariablesProvider.getEnvironmentVariables(copyContext);
 
 		AppDefinition definition =
-				new AppDefinition(String.format("%s:%s:%s",
-						taskExecution.getTaskName(),
-						workerStepExecution.getJobExecution().getJobInstance().getJobName(),
-						workerStepExecution.getStepName()),
-						this.environmentVariablesProvider.getEnvironmentVariables(copyContext));
+				new AppDefinition(resolveApplicationName(),
+						environmentVariables);
 
 		AppDeploymentRequest request =
 				new AppDeploymentRequest(definition,
@@ -241,6 +286,15 @@ public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAw
 		taskLauncher.launch(request);
 	}
 
+	private String resolveApplicationName() {
+		if(StringUtils.hasText(this.applicationName)) {
+			return this.applicationName;
+		}
+		else {
+			return this.taskExecution.getTaskName();
+		}
+	}
+
 	private String formatArgument(String key, String value) {
 		return String.format("--%s=%s", key, value);
 	}

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/NoOpEnvironmentVariablesProvider.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.cloud.task.batch.partition;
 
-import java.util.HashMap;
+import java.util.Collections;
 import java.util.Map;
 
 import org.springframework.batch.item.ExecutionContext;
@@ -38,6 +38,6 @@ public class NoOpEnvironmentVariablesProvider implements EnvironmentVariablesPro
 	 */
 	@Override
 	public Map<String, String> getEnvironmentVariables(ExecutionContext executionContext) {
-		return new HashMap<>(0);
+		return Collections.emptyMap();
 	}
 }

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/PassThroughCommandLineArgsProvider.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.List;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.util.Assert;
+
+/**
+ * Returns the {@code List<String>} provided.
+ *
+ * @author Michael Minella
+ * @since 1.1.0
+ */
+public class PassThroughCommandLineArgsProvider implements CommandLineArgsProvider {
+
+	private final List<String> commandLineArgs;
+
+	public PassThroughCommandLineArgsProvider(List<String> commandLineArgs) {
+		Assert.notNull(commandLineArgs, "commandLineArgs is required");
+
+		this.commandLineArgs = commandLineArgs;
+	}
+
+	@Override
+	public List<String> getCommandLineArgs(ExecutionContext executionContext) {
+		return commandLineArgs;
+	}
+}

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/SimpleCommandLineArgsProvider.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.List;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.cloud.task.repository.TaskExecution;
+import org.springframework.util.Assert;
+
+/**
+ * Returns any command line arguments used with the {@link TaskExecution} provided.
+ *
+ * @author Michael Minella
+ * @since 1.1.0
+ */
+public class SimpleCommandLineArgsProvider implements CommandLineArgsProvider {
+
+	private final TaskExecution taskExecution;
+
+	public SimpleCommandLineArgsProvider(TaskExecution taskExecution) {
+		Assert.notNull(taskExecution, "A taskExecution is required");
+
+		this.taskExecution = taskExecution;
+	}
+
+	@Override
+	public List<String> getCommandLineArgs(ExecutionContext executionContext) {
+		return this.taskExecution.getArguments();
+	}
+}

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/partition/SimpleEnvironmentVariablesProvider.java

@@ -28,7 +28,8 @@ import org.springframework.core.env.MapPropertySource;
 import org.springframework.core.env.PropertySource;
 
 /**
- * Copies all existing environment variables as made available in the {@link Environment}.
+ * Copies all existing environment variables as made available in the {@link Environment}
+ * only if includeCurrentEnvironment is set to true (default).
  * The <code>environmentProperties</code> option provides the ability to override any
  * specific values on an as needed basis.
  *
@@ -42,6 +43,8 @@ public class SimpleEnvironmentVariablesProvider implements EnvironmentVariablesP
 
 	private Map<String, String> environmentProperties = new HashMap<>(0);
 
+	private boolean includeCurrentEnvironment = true;
+
 	/**
 	 * @param environment The {@link Environment} for this context
 	 */
@@ -57,11 +60,24 @@ public class SimpleEnvironmentVariablesProvider implements EnvironmentVariablesP
 		this.environmentProperties = environmentProperties;
 	}
 
+	/**
+	 * Establishes if current environment variables will be included as a part of the provider.
+	 * @param includeCurrentEnvironment true(default) include local environment properties.  False do not include
+	 * current environment properties.
+	 */
+	public void setIncludeCurrentEnvironment(boolean includeCurrentEnvironment) {
+		this.includeCurrentEnvironment = includeCurrentEnvironment;
+	}
+
 	@Override
 	public Map<String, String> getEnvironmentVariables(ExecutionContext executionContext) {
 
 		Map<String, String> environmentProperties = new HashMap<>(this.environmentProperties.size());
-		environmentProperties.putAll(getCurrentEnvironmentProperties());
+
+		if(includeCurrentEnvironment) {
+			environmentProperties.putAll(getCurrentEnvironmentProperties());
+		}
+
 		environmentProperties.putAll(this.environmentProperties);
 
 		return environmentProperties;

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

@@ -149,7 +149,7 @@ public class DeployerPartitionHandlerTests {
 
 		AppDefinition appDefinition = request.getDefinition();
 
-		assertEquals("partitionedJobTask:partitionedJob:step1:partition1", appDefinition.getName());
+		assertEquals("partitionedJobTask", appDefinition.getName());
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID, "1")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID, "4")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")));
@@ -362,7 +362,7 @@ public class DeployerPartitionHandlerTests {
 
 		AppDefinition appDefinition = request.getDefinition();
 
-		assertEquals("partitionedJobTask:partitionedJob:step1:partition1", appDefinition.getName());
+		assertEquals("partitionedJobTask", appDefinition.getName());
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID, "1")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID, "4")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")));
@@ -426,7 +426,7 @@ public class DeployerPartitionHandlerTests {
 
 		AppDefinition appDefinition = request.getDefinition();
 
-		assertEquals("partitionedJobTask:partitionedJob:step1:partition1", appDefinition.getName());
+		assertEquals("partitionedJobTask", appDefinition.getName());
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID, "1")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID, "4")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")));
@@ -608,7 +608,7 @@ public class DeployerPartitionHandlerTests {
 
 		AppDefinition appDefinition = request.getDefinition();
 
-		assertEquals("partitionedJobTask:partitionedJob:step1:partition1", appDefinition.getName());
+		assertEquals("partitionedJobTask", appDefinition.getName());
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID, "1")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID, "4")));
 		assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")));
@@ -661,7 +661,27 @@ public class DeployerPartitionHandlerTests {
 		Collections.sort(allRequests, new Comparator<AppDeploymentRequest>() {
 			@Override
 			public int compare(AppDeploymentRequest o1, AppDeploymentRequest o2) {
-				return o1.getDefinition().getName().compareTo(o2.getDefinition().getName());
+				List<String> commandlineArguments = o1.getCommandlineArguments();
+
+				String o1Command = "";
+				for (String commandlineArgument : commandlineArguments) {
+					if(commandlineArgument.contains(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)) {
+						o1Command = commandlineArgument;
+						break;
+					}
+				}
+
+				commandlineArguments = o2.getCommandlineArguments();
+
+				String o2Command = "";
+				for (String commandlineArgument : commandlineArguments) {
+					if(commandlineArgument.contains(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID)) {
+						o2Command = commandlineArgument;
+						break;
+					}
+				}
+
+				return o1Command.compareTo(o2Command);
 			}
 		});
 
@@ -671,7 +691,7 @@ public class DeployerPartitionHandlerTests {
 			assertEquals(0, request.getDeploymentProperties().size());
 
 			AppDefinition appDefinition = request.getDefinition();
-			assertEquals("partitionedJobTask:partitionedJob:step1:partition" + (i - 3), appDefinition.getName());
+			assertEquals("partitionedJobTask", appDefinition.getName());
 			assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID, "1")));
 			assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID, String.valueOf(i))));
 			assertTrue(request.getCommandlineArguments().contains(formatArgs(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME, "step1")));

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/NoOpEnvironmentVariablesProviderTests.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ * @author Michael Minella
+ */
+public class NoOpEnvironmentVariablesProviderTests {
+
+	private NoOpEnvironmentVariablesProvider provider;
+
+	@Before
+	public void setUp() {
+		this.provider = new NoOpEnvironmentVariablesProvider();
+	}
+
+	@Test
+	public void test() {
+		Map<String, String> environmentVariables = this.provider.getEnvironmentVariables(null);
+		assertNotNull(environmentVariables);
+		assertTrue(environmentVariables.isEmpty());
+
+		Map<String, String> environmentVariables2 = this.provider.getEnvironmentVariables(null);
+		assertTrue(environmentVariables == environmentVariables2);
+	}
+}

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/PassThroughCommandLineArgsProviderTests.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ * @author Michael Minella
+ */
+public class PassThroughCommandLineArgsProviderTests {
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testNull() {
+		new PassThroughCommandLineArgsProvider(null);
+	}
+
+	@Test
+	public void test() {
+		List<String> args = Arrays.asList("foo", "bar", "baz");
+
+		CommandLineArgsProvider provider = new PassThroughCommandLineArgsProvider(args);
+
+		List<String> commandLineArgs = provider.getCommandLineArgs(null);
+
+		assertEquals("foo", commandLineArgs.get(0));
+		assertEquals("bar", commandLineArgs.get(1));
+		assertEquals("baz", commandLineArgs.get(2));
+	}
+}

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/partition/SimpleCommandLineArgsProviderTests.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2016 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.
+ * 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.springframework.cloud.task.batch.partition;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.springframework.cloud.task.repository.TaskExecution;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Michael Minella
+ */
+public class SimpleCommandLineArgsProviderTests {
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testNullConstructorArg() {
+		new SimpleCommandLineArgsProvider(null);
+	}
+
+	@Test
+	public void test() {
+		TaskExecution taskExecution = new TaskExecution();
+		taskExecution.setArguments(Arrays.asList("foo", "bar", "baz"));
+
+		CommandLineArgsProvider provider = new SimpleCommandLineArgsProvider(taskExecution);
+
+		List<String> commandLineArgs = provider.getCommandLineArgs(null);
+
+		assertEquals("foo", commandLineArgs.get(0));
+		assertEquals("bar", commandLineArgs.get(1));
+		assertEquals("baz", commandLineArgs.get(2));
+	}
+}