Scheduled Task does not recognize multiple profiles
@chrisjs commented on [Wed Apr 24 2019](https://github.com/spring-cloud/spring-cloud-dataflow-ui/issues/1178)

@jsa4000 commented on [Wed Apr 24 2019](https://github.com/spring-cloud/spring-cloud-dataflow/issues/3187)

**Description:**
Creating a scheduled task using kubernetes Spring Cloud Dataflow server, it does not parse correctly the arguments with commas. for example `--spring.profiles.active=k8s,master`. It seems to interprete the profile composition as two arguments.

`2019-04-24 07:38:12.456 DEBUG 1 --- [           main] o.s.c.t.r.support.SimpleTaskRepository   : Creating: TaskExecution{executionId=0, parentExecutionId=null, exitCode=null, taskName='batch-uploader-task', startTime=Wed Apr 24 07:38:12 GMT 2019, endTime=null, exitMessage='null', externalExecutionId='null', errorMessage='null', arguments=[--spring.datasource.username=postgres, --spring.datasource.url=jdbc:postgresql://eks-lab-dev-db.cwekrnapay4v.eu-west-2.rds.amazonaws.com:5432/dataflow, --spring.datasource.driverClassName=org.postgresql.Driver, --spring.cloud.task.name=batch-uploader-task, --version=0.0.1, --spring.datasource.password=password, --spring.profiles.active=k8s, master]}`

In this case the second profile is not recognized for the application.

This happens only creating scheduled tasks, since launching the tasks normally it work as expected.

**Release versions:**
Used docker version from docker hub.

- image: springcloud/spring-cloud-dataflow-server
- version: 2.0.2.RELEASE

**Custom apps:**
A custom App with a partitioner that needs two profiles to work.

**Steps to reproduce:**

1. Register the app

`app register --type task --name batch-uploader-app --uri docker:jsa4000/dataflow-batch-uploader-k8s:0.0.1-SNAPSHOT`

2. Create the task

`task create batch-uploader-task --definition "batch-uploader-app --version=0.0.1"`

3. Create a schedule task:

Task Name: `batch-uploader-task`
Schedule name: `myschedule`
Cron Expression: `*/1 * * * *`
Arguments:  `--spring.profiles.active=k8s,master` or `--spring.profiles.active="k8s,master"`



---

@sabbyanandan commented on [Wed Apr 24 2019](https://github.com/spring-cloud/spring-cloud-dataflow/issues/3187#issuecomment-486232483)

@jsa4000: Thanks for the report. That seems like a disconnect between a manual launch vs. launching on a recurring schedule. We will have a look. 

---

@chrisjs commented on [Wed Apr 24 2019](https://github.com/spring-cloud/spring-cloud-dataflow/issues/3187#issuecomment-486320460)

when `TaskSchedulerController.save` is called, `arguments` contain multiple entries, based on how many `,`'s it splits on, rather than a single value



---

@oodamien commented on [Mon Apr 29 2019](https://github.com/spring-cloud/spring-cloud-dataflow-ui/issues/1178#issuecomment-487611983)

The UI send the args as a string: 
`--foo1=bar1, bar2, --foo2=bar3`



diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerController.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerController.java
index 284cf08a4..4343dab82 100644
--- a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerController.java
+++ b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerController.java
@@ -136,9 +136,10 @@ public class TaskSchedulerController {
 	public void save(@RequestParam("scheduleName") String scheduleName,
 			@RequestParam("taskDefinitionName") String taskDefinitionName,
 			@RequestParam String properties,
-			@RequestParam(required = false) List<String> arguments) {
+			@RequestParam(required = false) String arguments) {
 		Map<String, String> propertiesToUse = DeploymentPropertiesUtils.parse(properties);
-		schedulerService.schedule(scheduleName, taskDefinitionName, propertiesToUse, arguments);
+		List<String> argumentsToUse = DeploymentPropertiesUtils.parseParamList(arguments, " ");
+		schedulerService.schedule(scheduleName, taskDefinitionName, propertiesToUse, argumentsToUse);
 	}
 
 	/**
diff --git a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerControllerTests.java b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerControllerTests.java
index 7035ddb5c..a05b8550c 100644
--- a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerControllerTests.java
+++ b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerControllerTests.java
@@ -191,15 +191,35 @@ public class TaskSchedulerControllerTests {
 
 	@Test
 	public void testCreateScheduleWithSensitiveFields() throws Exception {
-		AppRegistration registration = this.registry.save("testApp", ApplicationType.task,
-				"1.0.0", new URI("file:src/test/resources/apps/foo-task"), null);
+		String auditData = createScheduleWithArguments("argument1=foo password=secret");
+
+		assertEquals(
+				"{\\"commandlineArguments\\":[\\"argument1=foo\\",\\"password=******\\"],\\"taskDefinitionName\\":\\"testDefinition\\","
+						+ "\\"taskDefinitionProperties\\":{\\"prop1\\":\\"foo\\",\\"spring.datasource.username\\":null,\\"prop2.secret\\":\\"******\\",\\"spring.datasource.url\\":null,\\"spring.datasource.driverClassName\\":null,\\"spring.cloud.task.name\\":\\"testDefinition\\"},"
+						+ "\\"deploymentProperties\\":{\\"spring.cloud.deployer.prop1.secret\\":\\"******\\",\\"spring.cloud.deployer.prop2.password\\":\\"******\\"}}",
+				auditData);
+	}
+
+	@Test
+	public void testCreateScheduleCommaDelimitedArgs() throws Exception {
+		String auditData = createScheduleWithArguments("argument1=foo spring.profiles.active=k8s,master argument3=bar");
+
+		assertEquals(
+				"{\\"commandlineArguments\\":[\\"argument1=foo\\",\\"spring.profiles.active=k8s,master\\",\\"argument3=bar\\"],\\"taskDefinitionName\\":\\"testDefinition\\","
+						+ "\\"taskDefinitionProperties\\":{\\"prop1\\":\\"foo\\",\\"spring.datasource.username\\":null,\\"prop2.secret\\":\\"******\\",\\"spring.datasource.url\\":null,\\"spring.datasource.driverClassName\\":null,\\"spring.cloud.task.name\\":\\"testDefinition\\"},"
+						+ "\\"deploymentProperties\\":{\\"spring.cloud.deployer.prop1.secret\\":\\"******\\",\\"spring.cloud.deployer.prop2.password\\":\\"******\\"}}",
+				auditData);
+	}
+
+	private String createScheduleWithArguments(String arguments) throws Exception {
+		this.registry.save("testApp", ApplicationType.task, "1.0.0", new URI("file:src/test/resources/apps/foo-task"), null);
 
 		repository.save(new TaskDefinition("testDefinition", "testApp"));
 		mockMvc.perform(post("/tasks/schedules/").param("taskDefinitionName", "testDefinition")
 				.param("scheduleName", "mySchedule")
 				.param("properties",
 						"scheduler.cron.expression=* * * * *,app.testApp.prop1=foo,app.testApp.prop2.secret=kenny,deployer.*.prop1.secret=cartman,deployer.*.prop2.password=kyle")
-				.param("arguments", "argument1=foo,password=secret")
+				.param("arguments", arguments)
 				.accept(MediaType.APPLICATION_JSON)).andDo(print()).andExpect(status().isCreated());
 		assertEquals(1, simpleTestScheduler.list().size());
 		ScheduleInfo scheduleInfo = simpleTestScheduler.list().get(0);
@@ -216,12 +236,7 @@ public class TaskSchedulerControllerTests {
 		assertEquals(AuditActionType.CREATE, auditRecord.getAuditAction());
 		assertEquals("mySchedule", auditRecord.getCorrelationId());
 
-		assertEquals(
-				"{\\"commandlineArguments\\":[\\"argument1=foo\\",\\"password=******\\"],\\"taskDefinitionName\\":\\"testDefinition\\","
-						+ "\\"taskDefinitionProperties\\":{\\"prop1\\":\\"foo\\",\\"spring.datasource.username\\":null,\\"prop2.secret\\":\\"******\\",\\"spring.datasource.url\\":null,\\"spring.datasource.driverClassName\\":null,\\"spring.cloud.task.name\\":\\"testDefinition\\"},"
-						+ "\\"deploymentProperties\\":{\\"spring.cloud.deployer.prop1.secret\\":\\"******\\",\\"spring.cloud.deployer.prop2.password\\":\\"******\\"}}",
-				auditRecord.getAuditData());
-
+		return auditRecord.getAuditData();
 	}
 
 	@Test
['spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerController.java', 'spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/TaskSchedulerControllerTests.java']