Command line arguments not passed to ScheduleRequest
When on the "Create Schedules" page in the dashboard, I can add a key and value to the "Task Arguments" but that value is not passed to the ScheduleRequest.commandLineArguments.

But If you create the task with say `timestamp --timestamp.format=YYYY` it will be mapped as a property in the AppDefinition so that is working, just a note..
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java
index c5ed9f8f0..976892edc 100644
--- a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java
+++ b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java
@@ -52,6 +52,7 @@ import org.springframework.util.StringUtils;
  * for Scheduling tasks.
  *
  * @author Glenn Renfro
+ * @author Chris Schaefer
  */
 public class DefaultSchedulerService implements SchedulerService {
 
@@ -133,9 +134,8 @@ public class DefaultSchedulerService implements SchedulerService {
 				appDeploymentProperties, whitelistProperties);
 		DeploymentPropertiesUtils.validateDeploymentProperties(taskDeploymentProperties);
 		taskDeploymentProperties = extractAndQualifySchedulerProperties(taskDeploymentProperties);
-		ScheduleRequest scheduleRequest = new ScheduleRequest(revisedDefinition,
-				taskDeploymentProperties,
-				deployerDeploymentProperties, scheduleName, getTaskResource(taskDefinitionName));
+		ScheduleRequest scheduleRequest = new ScheduleRequest(revisedDefinition, taskDeploymentProperties,
+				deployerDeploymentProperties, commandLineArgs, scheduleName, getTaskResource(taskDefinitionName));
 		this.scheduler.schedule(scheduleRequest);
 	}
 
@@ -203,7 +203,7 @@ public class DefaultSchedulerService implements SchedulerService {
 		return result;
 	}
 
-	private Resource getTaskResource(String taskDefinitionName) {
+	protected Resource getTaskResource(String taskDefinitionName) {
 		TaskDefinition taskDefinition = this.taskDefinitionRepository.findOne(taskDefinitionName);
 		if (taskDefinition == null) {
 			throw new NoSuchTaskDefinitionException(taskDefinitionName);
diff --git a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceTests.java b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceTests.java
index f2ddd3202..df1b3d7b3 100644
--- a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceTests.java
+++ b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceTests.java
@@ -18,6 +18,7 @@ package org.springframework.cloud.dataflow.server.service.impl;
 
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -27,15 +28,19 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
 import org.springframework.boot.autoconfigure.jdbc.EmbeddedDataSourceConfiguration;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.dataflow.configuration.metadata.ApplicationConfigurationMetadataResolver;
 import org.springframework.cloud.dataflow.core.ApplicationType;
 import org.springframework.cloud.dataflow.core.TaskDefinition;
 import org.springframework.cloud.dataflow.registry.AppRegistry;
+import org.springframework.cloud.dataflow.registry.AppRegistryCommon;
 import org.springframework.cloud.dataflow.registry.domain.AppRegistration;
 import org.springframework.cloud.dataflow.server.DockerValidatorProperties;
 import org.springframework.cloud.dataflow.server.config.apps.CommonApplicationProperties;
@@ -43,21 +48,26 @@ import org.springframework.cloud.dataflow.server.configuration.TaskServiceDepend
 import org.springframework.cloud.dataflow.server.repository.TaskDefinitionRepository;
 import org.springframework.cloud.dataflow.server.service.SchedulerService;
 import org.springframework.cloud.dataflow.server.service.SchedulerServiceProperties;
+import org.springframework.cloud.deployer.resource.docker.DockerResource;
 import org.springframework.cloud.scheduler.spi.core.CreateScheduleException;
 import org.springframework.cloud.scheduler.spi.core.ScheduleInfo;
+import org.springframework.cloud.scheduler.spi.core.ScheduleRequest;
 import org.springframework.cloud.scheduler.spi.core.Scheduler;
 import org.springframework.core.io.Resource;
+import org.springframework.core.io.ResourceLoader;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-
 @RunWith(SpringRunner.class)
 @SpringBootTest(classes = { EmbeddedDataSourceConfiguration.class, TaskServiceDependencies.class,
 		PropertyPlaceholderAutoConfiguration.class }, properties = {
@@ -255,6 +265,52 @@ public class DefaultSchedulerServiceTests {
 		assertThat(schedules.size()).isEqualTo(0);
 	}
 
+	@Test
+	@DirtiesContext
+	public void testScheduleWithCommandLineArguments() {
+		List<String> commandLineArguments = getCommandLineArguments(Arrays.asList("--myArg1", "--myArg2"));
+
+		assertNotNull("Command line arguments should not be null", commandLineArguments);
+		assertEquals("Invalid number of command line arguments", 2, commandLineArguments.size());
+		assertEquals("Invalid command line argument", "--myArg1", commandLineArguments.get(0));
+		assertEquals("Invalid command line argument", "--myArg2", commandLineArguments.get(1));
+	}
+
+	@Test
+	@DirtiesContext
+	public void testScheduleWithoutCommandLineArguments() {
+		List<String> commandLineArguments = getCommandLineArguments(null);
+
+		assertNotNull("Command line arguments should not be null", commandLineArguments);
+		assertEquals("Invalid number of command line arguments", 0, commandLineArguments.size());
+	}
+
+	private List<String> getCommandLineArguments(List<String> commandLineArguments) {
+		Scheduler mockScheduler = mock(TaskServiceDependencies.SimpleTestScheduler.class);
+		TaskDefinitionRepository mockTaskDefinitionRepository = mock(TaskDefinitionRepository.class);
+		AppRegistryCommon mockAppRegistryCommon = mock(AppRegistryCommon.class);
+
+		SchedulerService mockSchedulerService = new DefaultSchedulerService(mock(CommonApplicationProperties.class),
+				mockScheduler, mockTaskDefinitionRepository, mockAppRegistryCommon, mock(ResourceLoader.class),
+				mock(TaskConfigurationProperties.class), mock(DataSourceProperties.class), "uri",
+				mock(ApplicationConfigurationMetadataResolver.class), mock(SchedulerServiceProperties.class));
+
+		TaskDefinition taskDefinition = new TaskDefinition(BASE_DEFINITION_NAME, "timestamp");
+
+		when(mockTaskDefinitionRepository.findOne(BASE_DEFINITION_NAME)).thenReturn(taskDefinition);
+		when(mockAppRegistryCommon.find(taskDefinition.getRegisteredAppName(), ApplicationType.task))
+				.thenReturn(new AppRegistration());
+		when(((DefaultSchedulerService)mockSchedulerService).getTaskResource(BASE_DEFINITION_NAME))
+				.thenReturn(new DockerResource("springcloudtask/timestamp-task:latest"));
+
+		mockSchedulerService.schedule(BASE_SCHEDULE_NAME, BASE_DEFINITION_NAME, this.testProperties,
+				commandLineArguments);
+
+		ArgumentCaptor<ScheduleRequest> scheduleRequestArgumentCaptor = ArgumentCaptor.forClass(ScheduleRequest.class);
+		verify(mockScheduler).schedule(scheduleRequestArgumentCaptor.capture());
+
+		return scheduleRequestArgumentCaptor.getValue().getCommandlineArguments();
+	}
 
 	private void verifyScheduleExistsInScheduler(ScheduleInfo scheduleInfo) {
 		List<ScheduleInfo> scheduleInfos = ((TaskServiceDependencies.SimpleTestScheduler)simpleTestScheduler).getSchedules();
['spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java', 'spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceTests.java']