Repository: spring-cloud/spring-cloud-dataflow
Issue: DataFlow fails to schedule CTR

When dataflow creates a schedule for a CTR it fails because it the DefaultSchedulerService requests an AppRegistration for the CTR and the AppRegisty throws an exception.  This is because the AppRegistry does not know how to handle CTRs (this is expected).  So the DefaultSchedulerService needs to add code to support requesting the CTR App (not the definition) from the AppRegistry.

Changed files:
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

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 41a6a6c05..c5ed9f8f0 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
@@ -203,16 +203,27 @@ public class DefaultSchedulerService implements SchedulerService {
 		return result;
 	}
 
-	private Resource getTaskResource(String taskDefinitionName ) {
+	private Resource getTaskResource(String taskDefinitionName) {
 		TaskDefinition taskDefinition = this.taskDefinitionRepository.findOne(taskDefinitionName);
 		if (taskDefinition == null) {
 			throw new NoSuchTaskDefinitionException(taskDefinitionName);
 		}
-
-		AppRegistration appRegistration = this.registry.find(taskDefinition.getRegisteredAppName(),
-				ApplicationType.task);
+		AppRegistration appRegistration = null;
+		if (isComposedDefinition(taskDefinition.getDslText())) {
+			appRegistration = this.registry.find(taskConfigurationProperties.getComposedTaskRunnerName(),
+					ApplicationType.task);
+		}
+		else {
+			appRegistration = this.registry.find(taskDefinition.getRegisteredAppName(),
+					ApplicationType.task);
+		}
 		Assert.notNull(appRegistration, "Unknown task app: " + taskDefinition.getRegisteredAppName());
 		return this.registry.getAppResource(appRegistration);
 	}
 
+	private boolean isComposedDefinition(String dsl) {
+		Assert.hasText(dsl, "dsl must not be empty nor null");
+		TaskParser taskParser = new TaskParser("__dummy", dsl, true, true);
+		return taskParser.parse().isComposed();
+	}
 }
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 32d3f7eaa..114e2e331 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
@@ -73,6 +73,8 @@ public class DefaultSchedulerServiceTests {
 
 	private static final String BASE_DEFINITION_NAME = "myTaskDefinition";
 
+	private static final String CTR_DEFINITION_NAME= "myCtrDefinition";
+
 	@Autowired
 	private Scheduler simpleTestScheduler;
 
@@ -97,7 +99,10 @@ public class DefaultSchedulerServiceTests {
 	@Before
 	public void setup() throws Exception{
 		this.appRegistry.save("demo", ApplicationType.task, new URI("file:src/test/resources/apps/foo-task"), new URI("file:src/test/resources/apps/foo-task"));
+		this.appRegistry.save("demo2", ApplicationType.task, new URI("file:src/test/resources/apps/foo-task"), new URI("file:src/test/resources/apps/foo-task"));
+
 		taskDefinitionRepository.save(new TaskDefinition(BASE_DEFINITION_NAME, "demo"));
+		taskDefinitionRepository.save(new TaskDefinition(CTR_DEFINITION_NAME, "demo && demo2"));
 		initializeSuccessfulRegistry();
 
 		this.testProperties = new HashMap<>();
@@ -121,6 +126,13 @@ public class DefaultSchedulerServiceTests {
 		verifyScheduleExistsInScheduler(createScheduleInfo(BASE_SCHEDULE_NAME));
 	}
 
+	@Test
+	@DirtiesContext
+	public void testScheduleCTR(){
+		schedulerService.schedule(BASE_SCHEDULE_NAME, CTR_DEFINITION_NAME, this.testProperties, this.commandLineArgs);
+		verifyScheduleExistsInScheduler(createScheduleInfo(BASE_SCHEDULE_NAME, CTR_DEFINITION_NAME));
+	}
+
 	@Test(expected = CreateScheduleException.class)
 	@DirtiesContext
 	public void testDuplicate(){
@@ -266,9 +278,13 @@ public class DefaultSchedulerServiceTests {
 	}
 
 	private ScheduleInfo createScheduleInfo(String scheduleName) {
+		return createScheduleInfo(scheduleName, BASE_DEFINITION_NAME);
+	}
+
+	private ScheduleInfo createScheduleInfo(String scheduleName, String taskDefinitionName) {
 		ScheduleInfo scheduleInfo = new ScheduleInfo();
 		scheduleInfo.setScheduleName(scheduleName);
-		scheduleInfo.setTaskDefinitionName(BASE_DEFINITION_NAME);
+		scheduleInfo.setTaskDefinitionName(taskDefinitionName);
 		scheduleInfo.setScheduleProperties(this.resolvedProperties);
 		return scheduleInfo;
 	}