Migrate Kubernetes Schedules from 2.3/2.4 to SCDF 2.5

Resolves #133

dataflow-migrate-schedules/README.adoc

@@ -1,14 +1,14 @@
-= Cloud Scheduler Migration
+= Spring Cloud Data Flow Schedule Migration
 
-The purpose of this project is to migrate existing schedules created with Spring
+The purpose of this project is to migrate existing schedules that are using the SchedulerTaskLauncher
-Cloud Data Flow 2.2.x and before to the new 2.3.0 format and stage the
+created with Spring Cloud Data Flow 2.3.0.RELEASE and 2.4.0.RELEASE to the new format.   This new format allows the platform scheduler
-SchedulerTaskLauncher.  This is a Spring Boot application that utilizes Spring Batch to create a workflow
+to launch the task directly, returning it to the functionality that existing prior to 2.3.0.RELEASE.
+This is a Spring Boot application that utilizes Spring Batch to create a workflow
 to migrate the schedules.  This is a single step Spring Batch Job that does the following:
 
 * Read - Retrieves all schedules from scheduler.
 
-* Process - Enriches the Schedule request with App and Deployer properties from the scheduler (or deployed app)
+* Process - Extracts the app properties from the SchedulerTaskLauncher instance and the associated platform job.
-as well as data from the TaskDefinition.
 
 * Write - Deploys artifacts if required and creates the new schedule.  Once the migrated
 schedule has been created, the old schedule is destroyed.
@@ -19,7 +19,7 @@ application and it will pick up where it left off.   Thanks Spring Batch! :-)
 == Build the project
 
 === Services Required
-In order to migrate the existing schedules to the 2.3.x Spring Cloud Data Flow Scheduling format, the Schedule Migrator requires the following services:
+In order to migrate the existing schedules, the Schedule Migrator requires the following services:
 
 1. Access to the database where Spring Cloud Data Flow stores its Task Definitions.  For Cloud Foundry we need to bind the database to the Schedule Migrator.
 2. Access to the Scheduling Agent.  For Cloud Foundry we need to bind the PCF Scheduler to the Scheduler Migrator.
@@ -34,7 +34,7 @@ In order to migrate the existing schedules to the 2.3.x Spring Cloud Data Flow S
 
 === Prerequisites
 
-Spring Cloud Data Flow 2.3+ must be installed and running prior to launching the Cloud Scheduler Migration app.
+Spring Cloud Data Flow 2.3+ must be installed and running prior to launching the migration app.
 
 === Launching your migration
 1) Create a manifest.yml file in a work directory.
@@ -57,9 +57,7 @@ applications:
     spring_cloud_deployer_cloudfoundry_skipSslValidation: <true/false>
     spring_cloud_deployer_cloudfoundry_services: <your SCDF database service,your scheduler service>
     spring_cloud_scheduler_cloudfoundry_schedulerUrl: <URL to scheduler service>
-    spring_profiles_active: cf
+    spring_profiles_active: cloudfoundry
-    spring.cloud.deployer.cloudfoundry.healthCheckTimeout: 300
-    spring.cloud.deployer.cloudfoundry.apiTimeout: 300
     dataflowServerUri: <Your SCDF version 2.3+ server URI>
     spring_cloud_task_closecontextEnabled: true
     remoteRepositories_repo1_url: https://repo.spring.io/libs-snapshot
@@ -87,12 +85,13 @@ From the `dataflow-migrate-schedules` directory launch the `runMigration.sh` usi
 chmod +x scripts/runMigration.sh
 ./scripts/runMigration.sh
 ```
+
 === Picking which schedules to migrate
 Use the `scheduleNamesToMigrate` property to specify a comma delimited list of
 the schedules you wish to migrate.  If you don't specify this property
 all schedules will be migrated.  For example:
 ```
-./scripts/runMigration.sh --scheduleNamesToMigrate=task-job3,task-job1
+./scripts/runMigration.sh --scheduleNamesToMigrate=schedulename-scdf-taskdefname,anotherschedulename-scdf-taskdefname
 ```
 
 === Limiting one Scheduler to run at a time
@@ -102,24 +101,88 @@ To enable this feature use the `runMigration.sh` script as follows.
 ./scripts/runMigration.sh --spring.cloud.task.single-instance-enabled=true
 ```
 
-=== Configuring Your Deployer Properties
-The following deployer properties will affect all schedules to be migrated.
-If a property is not set then the default will be used.
-
-==== Deployer properties to be applied to all migrated schedules:
-* healthCheckTimeout
-* apiTimeout
-* statusTimeout
-* stagingTimeout
-* startupTimeout
-* maximumConcurrentTasks
-* javaOpts
-
-NOTE: Descriptions of these properties can be found : https://github.com/cppwfs/spring-cloud-dataflow-samples/blob/SCDF-121/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java[here]
-
-=== Supported Databases
-The database supported are enumerated https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-rdbms[here].
-
 === Previously Pushed Apps
-The Cloud Schedule Migration app does not delete previously scheduled applications.
+The Cloud Schedule Migration app deletes SchedulerTaskLauncher and its associated schedule.
-If these apps are no longer needed it is up to the user to delete them.
+
+NOTE: If a single task definition has more than one schedule, the task definition
+droplet created for first migrated schedule will be used for all scheduled tasks of that task definition.
+
+== Running The Project For Kubernetes
+
+=== Prerequisites
+
+* Spring Cloud Data Flow 2.3.+ must be installed and running prior to launching the migration app.
+* Spring Cloud Data Flow must be using the `exec` entry point for scheduling apps.
+
+=== Launching your migration
+1) Establish the following environment variables:
+```
+export spring_datasource_url=<The Spring Cloud Data Flow database URL to be used by the migration tool>
+export spring_datasource_username=<The Spring Cloud Data Flow database username to be used by the migration tool>
+export spring_datasource_password=<The Spring Cloud Data Flow database password to be used by the migration tool>
+export spring_datasource_driverClassName=<The Spring Cloud Data Flow database driverClassName to be used by the migration tool>
+export KUBERNETES_NAMESPACE=< The namespace for the scheduled apps >
+export dbDriverClassName=<The Spring Cloud Data Flow database driverClassName to be used by the migrated tasks>
+export dbUrl=<The Spring Cloud Data Flow database database url to be used by the migrated tasks>
+export dbUserName=<The Spring Cloud Data Flow database user name to be used by the migrated tasks>
+export dbPassword=<The Spring Cloud Data Flow database password to be used by the migrated tasks>
+export dataflowUrl=<The URL to the current Spring Cloud Data Flow Server>
+```
+
+NOTE: There are two sets of datasource properties.  This is because the `spring_datasource_*`
+properties are used by the migration tool , while `db*` properties are used to
+set the database connection information that is required by the migrated tasks.
+The database connection URL used by the migration tool can be different than the
+one used by the tasks.
+
+2) Configure environment to access the cluster where the schedules are located.
+
+NOTE: In some cases if `KUBECONFIG` has a list of kubeconfig files the application may not select the proper kubeconfig file.
+In these cases set the `KUBECONFIG` so that it will use the proper  config file in the `$HOME/.kube/` directory.
+
+3) Be sure to establish a port forward to the database that Spring Cloud Data Flow is using,
+thus allowing the migration tool to gather information about task definitions.
+For example if using mysql: `kubectl port-forward <mysql pod name> 3306:3306`
+
+4) To start the migration:
+From the `dataflow-migrate-schedules` directory launch the `runKubernetesMigration.sh` using the commands below:
+```
+chmod +x scripts/runKubernetesMigration.sh
+./scripts/runKubernetesMigration.sh
+```
+
+=== Picking which schedules to migrate
+By default the `runKubernetesMigration.sh` will migrate all schedules.
+However if a specific set of schedules need to be migrated, then use the
+`scheduleNamesToMigrate` property to specify a comma delimited list of
+the schedules you wish to migrate.    For example:
+```
+./scripts/runKubernetesMigration.sh --scheduleNamesToMigrate=schedulename-scdf-taskdefname,anotherschedulename-scdf-taskdefname
+```
+
+=== Limiting one Scheduler to run at a time
+If there is a requirement that only one `schedulemigrator` should run at a time you can set the `spring.cloud.task.single-instance-enabled` property to true.   This will stop other executions of the schedulemigrator till the currently running instance completes.
+To enable this feature use the `runMigration.sh` script as follows.
+```
+./scripts/runKubernetesMigration.sh --spring.cloud.task.single-instance-enabled=true
+```
+
+== Configuring the Schedule Migration
+The following properties configure how the scheduler migrator will migrate the schedules.
+
+* schedulerToken - The token (default `scdf-`) is used by SchedulerTaskLauncher as a delimiter
+to separate the schedule name of each schedule into 2 components: `base schedule name`
+and `task name`.  This value will be used by the migration tool identify schedules to be migrated.
+* taskLauncherPrefix - The prefix used by the SchedulerTaskLauncher to mark the properties for the launched apps. Default: `tasklauncher`
+* scheduleNamesToMigrate - Comma delimited list of schedules to migrate.  If empty then all schedules will be migrated.
+* composedTaskRunnerRegisteredAppName - The registered application name for the composed task runner.  Default: `composed-task-runner`
+* dataflowUrl - The url of the Spring Cloud Data Flow Server that migrated composed task runners should execute task launch commands.
+
+=== Database Configuration for Kubernetes Migration
+* dbUserName - The user name of the database that contains the task definitions for schedules to be migrated.
+* dbPassword - The password of the database that contains the task definitions for schedules to be migrated.
+* dbUrl - The url to the database that contains the task definitions for schedules to be migrated.
+* dbDriverClassName - The driver class name to use for the database that contains the task definitions for schedules to be migrated.
+
+== Supported Databases
+The database supported are enumerated https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-rdbms[here].

dataflow-migrate-schedules/pom.xml

@@ -6,19 +6,19 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>2.2.1.RELEASE</version>
+		<version>2.2.6.RELEASE</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 	<groupId>io.spring</groupId>
 	<artifactId>migrateschedule</artifactId>
 	<version>1.0.0.BUILD-SNAPSHOT</version>
-	<name>Schedule Migrator</name>
+	<name>Schedule CD Migrator</name>
-	<description>Migrates SCDF Schedules to the 2.3 format</description>
+	<description>Migrates SCDF Schedules so that SchedulerTaskLauncher is no longer required.</description>
 
 	<properties>
 		<java.version>1.8</java.version>
-		<spring-cloud-data-flow.version>2.3.0.RC1</spring-cloud-data-flow.version>
+		<spring-cloud-data-flow.version>2.5.0.BUILD-SNAPSHOT</spring-cloud-data-flow.version>
-		<deployer.version>2.1.0.RELEASE</deployer.version>
+		<deployer.version>2.3.0.BUILD-SNAPSHOT</deployer.version>
 	</properties>
 
 	<dependencyManagement>
@@ -81,6 +81,10 @@
 		<dependency>
 			<groupId>org.mariadb.jdbc</groupId>
 			<artifactId>mariadb-java-client</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-jdbc</artifactId>
 		</dependency>
 		 <dependency>
 			 <groupId>org.springframework.cloud</groupId>
@@ -101,6 +105,11 @@
 			<artifactId>spring-cloud-dataflow-core</artifactId>
 			<version>${spring-cloud-data-flow.version}</version>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-dataflow-server-core</artifactId>
+			<version>${spring-cloud-data-flow.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-dataflow-registry</artifactId>
@@ -119,6 +128,11 @@
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-jdbc</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-deployer-kubernetes</artifactId>
+			<version>${deployer.version}</version>
+		</dependency>
 	</dependencies>
 
 	<build>

dataflow-migrate-schedules/scripts/runKubernetesMigration.sh

@@ -0,0 +1,4 @@
+export spring_profiles_active=kubernetes
+export spring_autoconfigure_exclude=org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryDeployerAutoConfiguration
+
+java -jar ./target/migrateschedule-1.0.0.BUILD-SNAPSHOT.jar $*

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/MigrateScheduleApplication.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -16,8 +16,11 @@
 
 package io.spring.migrateschedule;
 
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 public class MigrateScheduleApplication {
@@ -25,5 +28,4 @@ public class MigrateScheduleApplication {
 	public static void main(String[] args) {
 		SpringApplication.run(MigrateScheduleApplication.class, args);
 	}
-
 }

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -28,8 +28,8 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo;
 
 /**
  * Enriches the {@link ConvertScheduleInfo} with information obtained from the platform.
- * The new name for the schedule is established and the properties as well as commandline args
+ * The name for the schedule and the properties for the new schedule is
- * so that the SchedulerTaskLauncher can process the entries.
+ * extracted from the ScheduleTaskLauncher properties and command line args..
  *
  * @author Glenn Renfro
  */
@@ -48,7 +48,7 @@ public class SchedulerProcessor<T, C extends ScheduleInfo> implements ItemProces
 
 	@Override
 	public ConvertScheduleInfo process(ConvertScheduleInfo scheduleInfo){
-		if(scheduleInfo.getScheduleName().contains(migrateProperties.getSchedulerToken())) {
+		if(!scheduleInfo.getScheduleName().contains(migrateProperties.getSchedulerToken())) {
 			throw new ScheduleProcessedException(scheduleInfo.getScheduleName());
 		}
 		logger.info(String.format("Processing Schedule %s", scheduleInfo.getScheduleName()));

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerReader.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerWriter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -28,8 +28,7 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo;
 import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
 
 /**
- * Migrates the original schedule to the new scheduler format required for SCDF
+ * Migrates the SchedulerTaskLauncher schedule to the new schedule.
- * and stages the SchedulerTaskLauncher.
  *
  * @author Glenn Renfro
  */

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/BatchConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -47,8 +47,6 @@ import org.springframework.context.annotation.Configuration;
 @EnableBatchProcessing
 @EnableTask
 public class BatchConfiguration {
-
-
 	@Autowired
 	public JobBuilderFactory jobBuilderFactory;
 

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/CFMigrateScheduleConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -18,16 +18,15 @@ package io.spring.migrateschedule.configuration;
 
 import io.pivotal.reactor.scheduler.ReactorSchedulerClient;
 import io.pivotal.scheduler.SchedulerClient;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
 import io.spring.migrateschedule.service.CFMigrateSchedulerService;
 import io.spring.migrateschedule.service.MigrateProperties;
-import io.spring.migrateschedule.service.MigrateScheduleService;
 import io.spring.migrateschedule.service.TaskDefinitionRepository;
 import org.cloudfoundry.operations.CloudFoundryOperations;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
 import reactor.core.publisher.Mono;
 
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.domain.EntityScan;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
@@ -47,6 +46,7 @@ import org.springframework.context.annotation.Profile;
 @EntityScan({
 		"org.springframework.cloud.dataflow.core"
 })
+@Profile("cloudfoundry")
 public class CFMigrateScheduleConfiguration {
 
 	@Bean
@@ -69,9 +69,12 @@ public class CFMigrateScheduleConfiguration {
 	public CFMigrateSchedulerService scheduleService(CloudFoundryOperations cloudFoundryOperations,
 			SchedulerClient schedulerClient,
 			CloudFoundryConnectionProperties properties, MigrateProperties migrateProperties,
-			TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) {
+			TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties,
+			AppRegistrationRepository appRegistrationRepository,
+			TaskLauncher taskLauncher) {
 		return new CFMigrateSchedulerService(cloudFoundryOperations,
-				schedulerClient, properties, migrateProperties, taskDefinitionRepository, mavenProperties);
+				schedulerClient, properties, migrateProperties, taskDefinitionRepository,
+				mavenProperties, appRegistrationRepository, taskLauncher);
 	}
 
 	@Bean

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/KubernetesMigrateScheduleConfiguration.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2020 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 io.spring.migrateschedule.configuration;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
+import io.spring.migrateschedule.service.KubernetesMigrateSchedulerService;
+import io.spring.migrateschedule.service.MigrateProperties;
+import io.spring.migrateschedule.service.TaskDefinitionRepository;
+
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesClientFactory;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesDeployerProperties;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesScheduler;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesSchedulerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Profile;
+
+/**
+ * @author Glenn Renfro
+ */
+@Configuration
+@EntityScan({
+		"org.springframework.cloud.dataflow.core"
+})
+@Profile("kubernetes")
+public class KubernetesMigrateScheduleConfiguration {
+
+
+	@Bean
+	@Primary
+	public KubernetesSchedulerProperties kubernetesSchedulerProperties() {
+		return new KubernetesSchedulerProperties();
+	}
+
+
+	@Bean
+	public KubernetesClient kubernetesClient(KubernetesDeployerProperties kubernetesDeployerProperties) {
+		return KubernetesClientFactory.getKubernetesClient(kubernetesDeployerProperties);
+	}
+
+	@Bean
+	public KubernetesScheduler kubernetesScheduler(KubernetesClient kubernetesClient, KubernetesSchedulerProperties kubernetesSchedulerProperties) {
+		return new KubernetesScheduler( kubernetesClient, kubernetesSchedulerProperties);
+	}
+
+	@Bean
+	public KubernetesMigrateSchedulerService scheduleService( TaskDefinitionRepository taskDefinitionRepository, AppRegistrationRepository appRegistrationRepository, KubernetesClient kubernetesClient, MigrateProperties migrateProperties) {
+		return new KubernetesMigrateSchedulerService(taskDefinitionRepository, appRegistrationRepository, kubernetesClient, migrateProperties);
+	}
+
+}

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AbstractMigrateService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -18,14 +18,15 @@ package io.spring.migrateschedule.service;
 
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 
+import org.springframework.cloud.dataflow.core.AppRegistration;
+import org.springframework.cloud.dataflow.core.ApplicationType;
 import org.springframework.cloud.dataflow.core.TaskDefinition;
+import org.springframework.cloud.dataflow.core.dsl.TaskNode;
+import org.springframework.cloud.dataflow.core.dsl.TaskParser;
 import org.springframework.cloud.dataflow.registry.support.AppResourceCommon;
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
 import org.springframework.core.io.DefaultResourceLoader;
@@ -40,24 +41,48 @@ import org.springframework.util.StringUtils;
  */
 public abstract class AbstractMigrateService implements MigrateScheduleService {
 
-	private final static String DATA_FLOW_URI_KEY = "spring.cloud.dataflow.client.serverUri";
+	public final static String COMMAND_ARGUMENT_PREFIX = "cmdarg.";
 
-	private final static String COMMAND_ARGUMENT_PREFIX = "cmdarg.";
+	public final static String APP_PREFIX = "app.";
 
-	protected final static String APP_PREFIX = "app.";
+	public final static String DEPLOYER_PREFIX = "deployer.";
-
-	protected final static String DEPLOYER_PREFIX = "deployer.";
 
 	protected MigrateProperties migrateProperties;
 
-	private TaskDefinitionRepository taskDefinitionRepository;
+	protected TaskDefinitionRepository taskDefinitionRepository;
 
-	private MavenProperties mavenProperties;
+	protected AppRegistrationRepository appRegistrationRepository;
 
-	public AbstractMigrateService(MigrateProperties migrateProperties, TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) {
+	protected MavenProperties mavenProperties;
+
+	protected final String appArgPrefix;
+
+	protected final String deployerArgPrefix;
+
+	protected final String schedulerArgAppPrefix;
+
+	protected final String commandArgPrefix;
+
+	protected final int appArgPrefixLength;
+
+	protected final int deployerArgPrefixLength;
+
+	protected final int commandArgPrefixLength;
+
+	public AbstractMigrateService(MigrateProperties migrateProperties,
+			TaskDefinitionRepository taskDefinitionRepository,
+			MavenProperties mavenProperties, AppRegistrationRepository appRegistrationRepository) {
 		this.migrateProperties = migrateProperties;
 		this.taskDefinitionRepository = taskDefinitionRepository;
 		this.mavenProperties = mavenProperties;
+		this.appRegistrationRepository = appRegistrationRepository;
+		this.schedulerArgAppPrefix = String.format("--%s.", this.migrateProperties.getTaskLauncherPrefix());
+		this.appArgPrefix = String.format("%s%s", this.schedulerArgAppPrefix, APP_PREFIX);
+		this.deployerArgPrefix = String.format("%s%s", this.schedulerArgAppPrefix, DEPLOYER_PREFIX);
+		this.appArgPrefixLength = this.appArgPrefix.length();
+		this.deployerArgPrefixLength = this.deployerArgPrefix.length();
+		this.commandArgPrefix = String.format("%s%s.", COMMAND_ARGUMENT_PREFIX, this.migrateProperties.getTaskLauncherPrefix());
+		this.commandArgPrefixLength = this.commandArgPrefix.length();
 	}
 
 	public TaskDefinition findTaskDefinitionByName(String taskDefinitionName) {
@@ -76,7 +101,7 @@ public abstract class AbstractMigrateService implements MigrateScheduleService {
 	 * @param input the scheduler properties
 	 * @return scheduler properties for the task
 	 */
-	protected static Map<String, String> extractAndQualifySchedulerProperties(Map<String, String> input) {
+	protected Map<String, String> extractAndQualifySchedulerProperties(Map<String, String> input) {
 		final String prefix = "spring.cloud.scheduler.";
 
 		Map<String, String> result = new TreeMap<>(input).entrySet().stream()
@@ -88,83 +113,75 @@ public abstract class AbstractMigrateService implements MigrateScheduleService {
 	}
 
 	/**
-	 * Retrieve the resource for the SchedulerTaskLauncher and verify the URI.
+	 * Retrieve the resource for the Scheduled task
-	 * @return {@link Resource} for the SchedulerTaskLauncher.
+	 * @return {@link Resource} for the scheduled task.
 	 */
-	protected Resource getTaskLauncherResource() {
+	protected Resource getTaskLauncherResource(String resource) {
-		final URI url;
 		try {
-			new URI(this.migrateProperties.getSchedulerTaskLauncherUrl()); //verify url
+			new URI(resource); //verify url
 		}
 		catch (URISyntaxException uriSyntaxException) {
 			throw new IllegalArgumentException(uriSyntaxException);
 		}
 		AppResourceCommon appResourceCommon = new AppResourceCommon(this.mavenProperties, new DefaultResourceLoader());
-		return appResourceCommon.getResource(this.migrateProperties.getSchedulerTaskLauncherUrl());
+		return appResourceCommon.getResource(resource);
 	}
 
-	/**
+	protected String extractDeployerKey(String arg, int deployerPrefixLength) {
-	 * Add the appropriate tags to the command line args so that the SchedulerTaskLauncher can
+		return "spring.cloud.deployer." + extractAppKey(arg, deployerPrefixLength);
-	 * extract them.
+	}
-	 * @param args the command line args to be tagged.
-	 * @return the tagged command line args.
-	 */
-	protected List<String> tagCommandLineArgs(List<String> args) {
-		List<String> taggedArgs = new ArrayList<>();
 
-		for(String arg : args) {
+	protected String extractAppKey(String arg, int prefixLength) {
-			if(arg.contains("spring.cloud.task.name")) {
+		int indexOfEqual = arg.indexOf("=");
-				continue;
+		arg = arg.substring(prefixLength, indexOfEqual);
-			}
+		int dotIndex = arg.indexOf(".");
-			String updatedArg = arg;
+		String result = arg;
-			if (!arg.startsWith(DATA_FLOW_URI_KEY) && !"--".concat(arg).startsWith(DATA_FLOW_URI_KEY)) {
+		if (!arg.substring(0, dotIndex).equals("management")) {
-				updatedArg = COMMAND_ARGUMENT_PREFIX +
+			result = arg.substring(dotIndex + 1);
-						this.migrateProperties.getTaskLauncherPrefix() + "." + arg;
-			}
-			taggedArgs.add(updatedArg);
 		}
-		return taggedArgs;
+		return result;
 	}
 
-	/**
-	 * Add the appropriate tags to the command line args so that the SchedulerTaskLauncher can
-	 * extract them.
-	 * @param appName the name of the application to be associated with the property
-	 * @param appProperties the properties to be tagged
-	 * @param prefix the prefix to mark the property as to be used by the SchedulerTaskLauncher.
-	 * @return the tagged command line args.
-	 */
-	protected Map<String, String> tagProperties(String appName, Map<String, String> appProperties, String prefix) {
-		Map<String, String> taggedAppProperties = new HashMap<>(appProperties.size());
 
-		for(String key : appProperties.keySet()) {
+	protected String extractValue(String arg) {
-			if(key.contains("spring.cloud.task.name")) {
+		int indexOfEqual = arg.indexOf("=");
-				continue;
+		return arg.substring(indexOfEqual + 1);
+	}
+
+	protected ConvertScheduleInfo populateTaskDefinitionData(String appName, ConvertScheduleInfo scheduleInfo) {
+		TaskDefinition taskDefinition = this.taskDefinitionRepository.findByTaskName(appName);
+
+		if (taskDefinition != null) {
+			TaskParser taskParser = new TaskParser(appName, taskDefinition.getDslText(), false, false);
+			String registeredAppName = taskDefinition.getRegisteredAppName();
+			TaskNode taskNode = taskParser.parse();
+			if (taskNode.isComposed()) {
+				registeredAppName = this.migrateProperties.getComposedTaskRunnerRegisteredAppName();
+				scheduleInfo.setCtrDSL(taskNode.toExecutableDSL());
+				scheduleInfo.setCTR(true);
 			}
-			String updatedKey = key;
+
-			if (!key.startsWith(DATA_FLOW_URI_KEY)) {
+			AppRegistration appRegistration = this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(registeredAppName, ApplicationType.task);
-				if (StringUtils.hasText(appName)) {
+
-					updatedKey = this.migrateProperties.getTaskLauncherPrefix() + "." +
+			scheduleInfo.setTaskResource(getTaskLauncherResource(appRegistration.getUri().toString()));
-							prefix + appName + "." + key;
+			scheduleInfo.setTaskDefinitionName(appName);
-				}
-				else {
-					updatedKey = this.migrateProperties.getTaskLauncherPrefix() + "." +
-							prefix + key;
-				}
-			}
-			taggedAppProperties.put(updatedKey, appProperties.get(key));
 		}
-		return taggedAppProperties;
+		return scheduleInfo;
 	}
 
-	/**
+	protected ConvertScheduleInfo addDBInfoToScheduleInfo(ConvertScheduleInfo scheduleInfo) {
-	 * Add the required SchedulerTaskLauncher properties.
+		if(StringUtils.hasText(this.migrateProperties.getDbUrl())) {
-	 * @param properties the map of properties in which to add the SchedulerTaskLauncher properties.
+			scheduleInfo.getAppProperties().put("spring.datasource.url", this.migrateProperties.getDbUrl());
-	 * @return updated properties.
+		}
-	 */
+		if(StringUtils.hasText(this.migrateProperties.getDbDriverClassName())) {
-	protected Map<String, String> addSchedulerAppProps(Map<String, String> properties) {
+			scheduleInfo.getAppProperties().put("spring.datasource.driverClassName", this.migrateProperties.getDbDriverClassName());
-		Map<String, String> appProperties = new HashMap<>(properties);
+		}
-		appProperties.put("spring.cloud.dataflow.client.serverUri", this.migrateProperties.getDataflowServerUri());
+		if(StringUtils.hasText(this.migrateProperties.getDbUserName())) {
-		return appProperties;
+			scheduleInfo.getAppProperties().put("spring.datasource.username", this.migrateProperties.getDbUserName());
+		}
+		if(StringUtils.hasText(this.migrateProperties.getDbPassword())) {
+			scheduleInfo.getAppProperties().put("spring.datasource.password", this.migrateProperties.getDbPassword());
+		}
+		return scheduleInfo;
 	}
 }

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AppRegistrationRepository.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020 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 io.spring.migrateschedule.service;
+
+import java.util.List;
+
+import org.springframework.cloud.dataflow.core.AppRegistration;
+import org.springframework.cloud.dataflow.core.ApplicationType;
+import org.springframework.data.keyvalue.repository.KeyValueRepository;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * Repository interface for retrieving instances of the the {@link AppRegistration} class.
+ * @author Glenn Renfro
+ */
+@Transactional
+public interface AppRegistrationRepository extends KeyValueRepository<AppRegistration, Long> {
+
+	AppRegistration findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(String name, ApplicationType type);
+
+}

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/CFMigrateSchedulerService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -41,7 +41,6 @@ import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.cloud.dataflow.core.TaskDefinition;
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
 import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties;
 import org.springframework.cloud.deployer.spi.core.AppDefinition;
@@ -49,6 +48,7 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest;
 import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
 import org.springframework.cloud.deployer.spi.scheduler.SchedulerException;
 import org.springframework.cloud.deployer.spi.scheduler.SchedulerPropertyKeys;
+import org.springframework.cloud.deployer.spi.task.TaskLauncher;
 import org.springframework.util.StringUtils;
 
 /**
@@ -61,6 +61,8 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 
 	public static final String JAR_LAUNCHER = "org.springframework.boot.loader.JarLauncher";
 
+	public static final String TASK_NAME_ARGUMENT_KEY = "--spring.cloud.scheduler.task.launcher.taskName=";
+
 	private static final int JAR_LAUNCHER_LENGTH = JAR_LAUNCHER.length();
 
 	private static final Logger logger = LoggerFactory.getLogger(CFMigrateSchedulerService.class);
@@ -77,15 +79,20 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 
 	private CloudFoundryConnectionProperties properties;
 
+	private TaskLauncher taskLauncher;
+
 
 	public CFMigrateSchedulerService(CloudFoundryOperations cloudFoundryOperations,
 			SchedulerClient schedulerClient,
 			CloudFoundryConnectionProperties properties, MigrateProperties migrateProperties,
-			TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) {
+			TaskDefinitionRepository taskDefinitionRepository,
-		super(migrateProperties, taskDefinitionRepository, mavenProperties);
+			MavenProperties mavenProperties, AppRegistrationRepository appRegistrationRepository,
+			TaskLauncher taskLauncher) {
+		super(migrateProperties, taskDefinitionRepository, mavenProperties, appRegistrationRepository);
 		this.cloudFoundryOperations = cloudFoundryOperations;
 		this.schedulerClient = schedulerClient;
 		this.properties = properties;
+		this.taskLauncher = taskLauncher;
 	}
 
 	@Override
@@ -93,9 +100,9 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 		List<ConvertScheduleInfo> result = new ArrayList<>();
 		int pageCount = getJobPageCount();
 		for (int i = PCF_PAGE_START_NUM; i <= pageCount; i++) {
-			logger.info(String.format("Reading Schedules Page %s of %s ", i, pageCount ));
+			logger.info(String.format("Reading Schedules Page %s of %s ", i, pageCount));
 			List<ConvertScheduleInfo> scheduleInfoPage = getSchedules(i);
-			if(scheduleInfoPage == null) {
+			if (scheduleInfoPage == null) {
 				throw new SchedulerException(SCHEDULER_SERVICE_ERROR_MESSAGE);
 			}
 			result.addAll(scheduleInfoPage);
@@ -154,8 +161,8 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 
 	private boolean isScheduleMigratable(String scheduleName) {
 		boolean result;
-		if(migrateProperties.getScheduleNamesToMigrate().size() > 0) {
+		if (this.migrateProperties.getScheduleNamesToMigrate().size() > 0) {
-			result = migrateProperties.getScheduleNamesToMigrate().contains(scheduleName);
+			result = this.migrateProperties.getScheduleNamesToMigrate().contains(scheduleName);
 		}
 		else {
 			result = true;
@@ -174,7 +181,7 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 
 		logger.info(String.format("Retrieving ApplicationManifest for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
 		ApplicationManifest applicationManifest = getApplicationManifest(scheduleInfo.getTaskDefinitionName());
-		if(applicationManifest != null) {
+		if (applicationManifest != null) {
 			addApplicationManifestPropsToConvertScheduleInfo(scheduleInfo, applicationManifest);
 		}
 		if (environment != null) {
@@ -183,52 +190,64 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 			}
 		}
 		logger.info(String.format("Tagging command line args for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
-		List<String> revisedCommandLineArgs = tagCommandLineArgs(scheduleInfo.getCommandLineArgs());
+
-		revisedCommandLineArgs.add("--spring.cloud.scheduler.task.launcher.taskName=" + scheduleInfo.getTaskDefinitionName());
+		String appName = getAppNameFromArgs(scheduleInfo);
-		scheduleInfo.setCommandLineArgs(revisedCommandLineArgs);
+
-		Map<String, String> appProperties = null;
+		Map<String, String> appProperties;
 		try {
 			logger.info(String.format("Extracting Spring App Properties for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
 			appProperties = getSpringAppProperties(scheduleInfo.getScheduleProperties());
-			if(appProperties.size() > 0) {
+			if (appProperties.size() > 0) {
 				scheduleInfo.setUseSpringApplicationJson(true);
 			}
 		}
 		catch (Exception exception) {
 			throw new IllegalArgumentException("Unable to parse SPRING_APPLICATION_JSON from USER VARIABLES", exception);
 		}
-		logger.info(String.format("Retrieving Task Definition for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
+		populateTaskDefinitionData(appName, scheduleInfo);
-		TaskDefinition taskDefinition = findTaskDefinitionByName(appProperties.get("spring.cloud.task.name"));
+
-		if (appProperties.size() > 0 && taskDefinition == null) {
+		appProperties = cleanseAppProperties(scheduleInfo, appProperties);
-			throw new IllegalStateException(String.format("The schedule %s contains " +
+		if (scheduleInfo.isCTR()) {
-							"properties but the task definition %s does not exist and thus can't be migrated",
+			appProperties.put("graph", scheduleInfo.getCtrDSL());
-					scheduleInfo.getScheduleName(), scheduleInfo.getTaskDefinitionName()));
 		}
+
+		logger.info(String.format("Retrieving Task Definition for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
+
 		logger.info(String.format("Tagging app properties for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName()));
-		appProperties = tagProperties(taskDefinition.getRegisteredAppName(), appProperties, APP_PREFIX);
+		Map<String, String> deployerProperties = getDeployerProperties(scheduleInfo);
-		Map<String, String> deployerProperties = tagProperties(taskDefinition.getRegisteredAppName(),
+		appProperties.put("dataflow-server-uri", this.migrateProperties.getDataflowUrl());
-				getDeployerProperties(scheduleInfo), DEPLOYER_PREFIX);
-		appProperties = addSchedulerAppProps(appProperties);
 		appProperties.putAll(deployerProperties);
 		scheduleInfo.setAppProperties(appProperties);
+
+		scheduleInfo.setCommandLineArgs(cleanseCommandLineArgs(scheduleInfo, appName));
+		addDBInfoToScheduleInfo(scheduleInfo);
 		return scheduleInfo;
 	}
 
 	@Override
 	public void migrateSchedule(Scheduler scheduler, ConvertScheduleInfo scheduleInfo) {
-		String scheduleName = scheduleInfo.getScheduleName() + "-" + getSchedulePrefixDefinitionName(scheduleInfo.getTaskDefinitionName());
+		logger.info("Migrating " + scheduleInfo);
-		AppDefinition appDefinition = new AppDefinition(scheduleName, scheduleInfo.getAppProperties());
+		if (scheduleInfo.getTaskResource() == null) {
+			logger.info(String.format("The task definition name for schedule %s does not exist in Data Flow",
+					scheduleInfo.getScheduleName()));
+			return;
+		}
+		String scheduleName = scheduleInfo.getScheduleName().substring(0,
+				scheduleInfo.getScheduleName().indexOf(this.migrateProperties.getSchedulerToken()) - 1);
+		AppDefinition appDefinition = new AppDefinition(scheduleInfo.getTaskDefinitionName(), scheduleInfo.getAppProperties());
 		logger.info(String.format("Extracting schedule specific properties for schedule %s", scheduleInfo.getScheduleName()));
 		Map<String, String> schedulerProperties = extractAndQualifySchedulerProperties(scheduleInfo.getScheduleProperties());
-		ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, new HashMap<>(), scheduleInfo.getCommandLineArgs(), scheduleName, getTaskLauncherResource());
+		ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, new HashMap<>(), scheduleInfo.getCommandLineArgs(), scheduleName, scheduleInfo.getTaskResource());
 		logger.info(String.format("Staging ScheduleTaskLauncher and scheduling %s", scheduleInfo.getScheduleName()));
+
 		scheduler.schedule(scheduleRequest);
 		logger.info(String.format("Unscheduling original %s", scheduleInfo.getScheduleName()));
-		scheduler.unschedule(scheduleInfo.getScheduleName());
+		this.taskLauncher.destroy(scheduleInfo.getScheduleName());
 	}
 
 	/**
 	 * Retrieves the number of pages that can be returned when retrieving a list of jobs.
+	 *
 	 * @return an int containing the number of available pages.
 	 */
 	private int getJobPageCount() {
@@ -237,21 +256,18 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 					.spaceId(requestSummary.getId())
 					.detailed(false).build());
 		}).block();
-		if(response == null) {
+		if (response == null) {
 			throw new SchedulerException(SCHEDULER_SERVICE_ERROR_MESSAGE);
 		}
 		return response.getPagination().getTotalPages();
 	}
 
 	private Map<String, String> getSpringAppProperties(Map<String, String> properties) throws Exception {
-		Map<String, String> result;
+		Map<String, String> result = new HashMap<>();
-		if(properties.containsKey("SPRING_APPLICATION_JSON")) {
+		if (properties.containsKey("SPRING_APPLICATION_JSON")) {
 			result = new ObjectMapper()
 					.readValue(properties.get("SPRING_APPLICATION_JSON"), Map.class);
 		}
-		else {
-			result = new HashMap<>();
-		}
 		return result;
 	}
 
@@ -307,11 +323,12 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 				.filter(application -> appId.equals(application.getId()))
 				.singleOrEmpty();
 	}
+
 	private ApplicationManifest getApplicationManifest(String appName) {
-			return this.cloudFoundryOperations.applications()
+		return this.cloudFoundryOperations.applications()
-					.getApplicationManifest(GetApplicationManifestRequest
+				.getApplicationManifest(GetApplicationManifestRequest
-							.builder().name(appName).build())
+						.builder().name(appName).build())
-					.block();
+				.block();
 	}
 
 	private Map<String, String> getDeployerProperties(ConvertScheduleInfo scheduleInfo) {
@@ -320,10 +337,10 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 			result.put(CLOUD_FOUNDRY_PREFIX + ".buildpack", scheduleInfo.getJavaBuildPack());
 		}
 		if (scheduleInfo.getMemoryInMB() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX +  ".memory", scheduleInfo.getMemoryInMB() + "m");
+			result.put(CLOUD_FOUNDRY_PREFIX + ".memory", scheduleInfo.getMemoryInMB() + "m");
 		}
 		if (scheduleInfo.getDiskInMB() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX +  ".disk", scheduleInfo.getDiskInMB() + "m");
+			result.put(CLOUD_FOUNDRY_PREFIX + ".disk", scheduleInfo.getDiskInMB() + "m");
 		}
 		if (scheduleInfo.getApplicationHealthCheck() != null) {
 			result.put(CLOUD_FOUNDRY_PREFIX + ".health-check", scheduleInfo.getApplicationHealthCheck().getValue());
@@ -344,45 +361,22 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 			result.put(CLOUD_FOUNDRY_PREFIX + ".host", StringUtils.arrayToCommaDelimitedString(scheduleInfo.getHosts().toArray()));
 		}
 
-		// Global deployer properties;
-		if (this.migrateProperties.getHealthCheckTimeout() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".health-check-timeout", this.migrateProperties.getHealthCheckTimeout());
-		}
-		if (this.migrateProperties.getJavaOptions() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".javaOpts", this.migrateProperties.getJavaOptions());
-		}
-		if (this.migrateProperties.getApiTimeout() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".api-timeout", String.valueOf(this.migrateProperties.getApiTimeout()));
-		}
-		if (this.migrateProperties.getStatusTimeout() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".status-timeout", String.valueOf(this.migrateProperties.getStatusTimeout()));
-		}
-		if (this.migrateProperties.getStagingTimeout() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".staging-timeout", String.valueOf(this.migrateProperties.getStagingTimeout()));
-		}
-		if (this.migrateProperties.getStartupTimeout() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".startup-timeout", String.valueOf(this.migrateProperties.getStartupTimeout()));
-		}
-		if (this.migrateProperties.getMaximumConcurrentTasks() != null) {
-			result.put(CLOUD_FOUNDRY_PREFIX + ".maximum-concurrent-tasks", String.valueOf(this.migrateProperties.getMaximumConcurrentTasks()));
-		}
-
 		return result;
 	}
 
-	private ConvertScheduleInfo addApplicationManifestPropsToConvertScheduleInfo(ConvertScheduleInfo scheduleInfo, ApplicationManifest applicationManifest) {
+	private void addApplicationManifestPropsToConvertScheduleInfo(ConvertScheduleInfo scheduleInfo, ApplicationManifest applicationManifest) {
 		scheduleInfo.setDiskInMB(applicationManifest.getDisk());
 		scheduleInfo.setMemoryInMB(applicationManifest.getMemory());
 		scheduleInfo.setApplicationHealthCheck(applicationManifest.getHealthCheckType());
 		scheduleInfo.setJavaBuildPack(applicationManifest.getBuildpack());
 		scheduleInfo.setHealthCheckEndPoint(applicationManifest.getHealthCheckHttpEndpoint());
-		if(applicationManifest.getServices() != null && applicationManifest.getServices().size() > 0) {
+		if (applicationManifest.getServices() != null && applicationManifest.getServices().size() > 0) {
 			scheduleInfo.setServices(applicationManifest.getServices());
 		}
-		if(applicationManifest.getDomains() != null && applicationManifest.getDomains().size() > 0) {
+		if (applicationManifest.getDomains() != null && applicationManifest.getDomains().size() > 0) {
 			scheduleInfo.setDomains(applicationManifest.getDomains());
 		}
-		if(applicationManifest.getRoutes() != null && applicationManifest.getRoutes().size() > 0) {
+		if (applicationManifest.getRoutes() != null && applicationManifest.getRoutes().size() > 0) {
 			List<String> routes = new ArrayList<>();
 			for (Route route : applicationManifest.getRoutes()) {
 				routes.add(route.getRoute());
@@ -390,13 +384,101 @@ public class CFMigrateSchedulerService extends AbstractMigrateService {
 			scheduleInfo.setRoutes(routes);
 		}
 		if (applicationManifest.getHosts() != null && applicationManifest.getHosts().size() > 0) {
-			List<String> hosts = new ArrayList<>();
+			List<String> hosts = new ArrayList<>(applicationManifest.getHosts());
-			for (String host : applicationManifest.getHosts()) {
-				hosts.add(host);
-			}
 			scheduleInfo.setHosts(hosts);
 		}
+	}
 
-		return scheduleInfo;
+	private List<String> cleanseCommandLineArgs(ConvertScheduleInfo scheduleInfo, String appName) {
+		List<String> commandLineArgs = new ArrayList<>();
+
+		for (String arg : scheduleInfo.getCommandLineArgs()) {
+			String resultArg = (arg.startsWith("--")) ? "--" : "";
+			if (arg.startsWith(COMMAND_ARGUMENT_PREFIX)) {
+				commandLineArgs.add(resultArg + extractAppKey(arg, this.appArgPrefixLength) + "=" + extractValue(arg));
+			}
+			else if (arg.startsWith(this.deployerArgPrefix)) {
+				commandLineArgs.add(resultArg + extractDeployerKey(arg, this.appArgPrefixLength) + "=" + extractValue(arg));
+			}
+			else if (arg.startsWith(this.commandArgPrefix)) {
+				commandLineArgs.add("--" + arg.substring(this.commandArgPrefixLength));
+			}
+			else {
+				if (arg.startsWith("--spring.cloud.scheduler.task.launcher.taskName")) {
+					continue;
+				}
+				commandLineArgs.add(arg);
+			}
+		}
+		commandLineArgs.add("--spring.application.name=" + appName);
+		return commandLineArgs;
+	}
+
+	private String cleanseAppPropertyKey(String key) {
+		String prefix = String.format("%s.%s", this.migrateProperties.getTaskLauncherPrefix(), APP_PREFIX);
+		key = key.substring(prefix.length());
+		int dotIndex = key.indexOf(".");
+		String result = key;
+		if (!key.substring(0, dotIndex).equals("management")) {
+			result = key.substring(dotIndex + 1);
+		}
+		return result;
+	}
+
+	private Map<String, String> cleanseAppProperties(ConvertScheduleInfo scheduleInfo, Map<String, String> properties) {
+		Map<String, String> result = new HashMap<>();
+		String ctrProperties = "";
+		boolean isFirstCTREntry = true;
+		for (String key : properties.keySet()) {
+			if (scheduleInfo.isCTR()) {
+				String ctrPropertyCandidate = prepScheduleForCTR(scheduleInfo, key, properties.get(key));
+				if (ctrPropertyCandidate != null) {
+					if (isFirstCTREntry) {
+						isFirstCTREntry = false;
+						ctrProperties += ctrPropertyCandidate;
+					}
+					else {
+						ctrProperties = ctrProperties + "," + ctrPropertyCandidate;
+					}
+					continue;
+				}
+			}
+			if (key.startsWith(this.migrateProperties.getTaskLauncherPrefix())) {
+				String cleansedKey = cleanseAppPropertyKey(key);
+				result.put(cleansedKey, properties.get(key));
+			}
+			else {
+				result.put(key, properties.get(key));
+			}
+		}
+		if (!ctrProperties.isEmpty()) {
+			result.put("composed-task-properties", ctrProperties);
+		}
+		return result;
+	}
+
+	private String prepScheduleForCTR(ConvertScheduleInfo scheduleInfo, String key, String value) {
+		String ctrProperty = null;
+		String prefix = String.format("%s.%s%s.", this.migrateProperties.getTaskLauncherPrefix(), APP_PREFIX, scheduleInfo.getTaskDefinitionName());
+		if (key.startsWith(prefix)) {
+			String newKey = key.substring(prefix.length());
+			String appName = newKey.substring(0, newKey.indexOf("."));
+			ctrProperty = String.format("%s%s-%s.%s%s=%s", APP_PREFIX,
+					scheduleInfo.getTaskDefinitionName(), appName, APP_PREFIX,
+					newKey, value);
+		}
+		return ctrProperty;
+	}
+
+	private String getAppNameFromArgs(ConvertScheduleInfo scheduleInfo) {
+		String appName = null;
+		for (String command : scheduleInfo.getCommandLineArgs()) {
+			if (command.startsWith(TASK_NAME_ARGUMENT_KEY)) {
+				int appNameIndex = command.indexOf(TASK_NAME_ARGUMENT_KEY);
+				appName = command.substring(appNameIndex + TASK_NAME_ARGUMENT_KEY.length());
+				break;
+			}
+		}
+		return appName;
 	}
 }

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ConvertScheduleInfo.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -24,6 +24,7 @@ import java.util.Map;
 import org.cloudfoundry.operations.applications.ApplicationHealthCheck;
 
 import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo;
+import org.springframework.core.io.Resource;
 
 /**
  * A child implementation of {@link ScheduleInfo} that adds additional attributes
@@ -39,6 +40,8 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 
 	private Map<String, String> appProperties = new HashMap<>();
 
+	private Map<String, String> deployerProperties = new HashMap();
+
 	private Integer diskInMB;
 
 	private Integer memoryInMB;
@@ -49,8 +52,6 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 
 	private String healthCheckEndPoint;
 
-	private Integer maximumConcurrentTasks = 20;
-
 	private boolean useSpringApplicationJson;
 
 	private List<String> services;
@@ -61,6 +62,12 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 
 	private List<String> hosts;
 
+	private Resource taskResource;
+
+	private boolean isCTR;
+
+	private String ctrDSL;
+
 	public List<String> getCommandLineArgs() {
 		return commandLineArgs;
 	}
@@ -70,7 +77,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public String getRegisteredAppName() {
-		return registeredAppName;
+		return this.registeredAppName;
 	}
 
 	public void setRegisteredAppName(String registeredAppName) {
@@ -94,7 +101,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public Integer getDiskInMB() {
-		return diskInMB;
+		return this.diskInMB;
 	}
 
 	public void setDiskInMB(Integer diskInMB) {
@@ -102,7 +109,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public Integer getMemoryInMB() {
-		return memoryInMB;
+		return this.memoryInMB;
 	}
 
 	public void setMemoryInMB(Integer memoryInMB) {
@@ -110,7 +117,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public String getJavaBuildPack() {
-		return javaBuildPack;
+		return this.javaBuildPack;
 	}
 
 	public void setJavaBuildPack(String javaBuildPack) {
@@ -118,7 +125,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public ApplicationHealthCheck getApplicationHealthCheck() {
-		return applicationHealthCheck;
+		return this.applicationHealthCheck;
 	}
 
 	public void setApplicationHealthCheck(ApplicationHealthCheck applicationHealthCheck) {
@@ -126,7 +133,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public boolean isUseSpringApplicationJson() {
-		return useSpringApplicationJson;
+		return this.useSpringApplicationJson;
 	}
 
 	public void setUseSpringApplicationJson(boolean useSpringApplicationJson) {
@@ -134,23 +141,15 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public String getHealthCheckEndPoint() {
-		return healthCheckEndPoint;
+		return this.healthCheckEndPoint;
 	}
 
 	public void setHealthCheckEndPoint(String healthCheckEndPoint) {
 		this.healthCheckEndPoint = healthCheckEndPoint;
 	}
 
-	public Integer getMaximumConcurrentTasks() {
-		return maximumConcurrentTasks;
-	}
-
-	public void setMaximumConcurrentTasks(Integer maximumConcurrentTasks) {
-		this.maximumConcurrentTasks = maximumConcurrentTasks;
-	}
-
 	public List<String> getServices() {
-		return services;
+		return this.services;
 	}
 
 	public void setServices(List<String> services) {
@@ -158,7 +157,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public List<String> getDomains() {
-		return domains;
+		return this.domains;
 	}
 
 	public void setDomains(List<String> domains) {
@@ -166,7 +165,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public List<String> getRoutes() {
-		return routes;
+		return this.routes;
 	}
 
 	public void setRoutes(List<String> routes) {
@@ -174,10 +173,42 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{
 	}
 
 	public List<String> getHosts() {
-		return hosts;
+		return this.hosts;
 	}
 
 	public void setHosts(List<String> hosts) {
 		this.hosts = hosts;
 	}
+
+	public Resource getTaskResource() {
+		return this.taskResource;
+	}
+
+	public void setTaskResource(Resource taskResource) {
+		this.taskResource = taskResource;
+	}
+
+	public boolean isCTR() {
+		return this.isCTR;
+	}
+
+	public void setCTR(boolean CTR) {
+		isCTR = CTR;
+	}
+
+	public String getCtrDSL() {
+		return this.ctrDSL;
+	}
+
+	public void setCtrDSL(String ctrDSL) {
+		this.ctrDSL = ctrDSL;
+	}
+
+	public Map<String, String> getDeployerProperties() {
+		return deployerProperties;
+	}
+
+	public void setDeployerProperties(Map<String, String> deployerProperties) {
+		this.deployerProperties = deployerProperties;
+	}
 }

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/KubernetesMigrateSchedulerService.java

@@ -0,0 +1,190 @@
+/*
+ * Copyright 2020 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 io.spring.migrateschedule.service;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.batch.CronJob;
+import io.fabric8.kubernetes.api.model.batch.CronJobList;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.dataflow.core.TaskDefinition;
+import org.springframework.cloud.deployer.resource.maven.MavenProperties;
+import org.springframework.cloud.deployer.spi.core.AppDefinition;
+import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest;
+import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
+
+
+/**
+ * Services required to migrate schedules to the 2.5.0 format in Kubernetes
+ * and stage the SchedulerTaskLauncher.
+ *
+ * @author Glenn Renfro
+ */
+public class KubernetesMigrateSchedulerService extends AbstractMigrateService {
+
+	private static final Logger logger = LoggerFactory.getLogger(KubernetesMigrateSchedulerService.class);
+
+	private KubernetesClient kubernetesClient;
+
+	public KubernetesMigrateSchedulerService(TaskDefinitionRepository taskDefinitionRepository,
+			AppRegistrationRepository appRegistrationRepository,
+			KubernetesClient kubernetesClient, MigrateProperties migrateProperties) {
+		super(migrateProperties, taskDefinitionRepository, new MavenProperties(), appRegistrationRepository);
+		this.kubernetesClient = kubernetesClient;
+		this.migrateProperties = migrateProperties;
+	}
+
+	/**
+	 * Retrieves all the cronjobs from the kubernetes instance and extracts
+	 * the schedule name, image  and command line args from each cronjob and
+	 * creates a {@link ConvertScheduleInfo}.
+	 * @return a {@link List} of {@link ConvertScheduleInfo}s.
+	 */
+	@Override
+	public List<ConvertScheduleInfo> scheduleInfoList() {
+		List<ConvertScheduleInfo> result = new ArrayList<>();
+		CronJobList cronJobs = this.kubernetesClient.batch().cronjobs().list();
+		for (CronJob cronjob : cronJobs.getItems()) {
+			Container container = cronjob.getSpec().getJobTemplate().getSpec().getTemplate().getSpec().getContainers().get(0);
+			ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo();
+			scheduleInfo.setScheduleName(cronjob.getMetadata().getName());
+			if (!isScheduleMigratable(scheduleInfo.getScheduleName())) {
+				continue;
+			}
+			String appName = scheduleInfo.getScheduleName().substring(
+					scheduleInfo.getScheduleName().indexOf(
+							this.migrateProperties.getSchedulerToken()) + this.migrateProperties.getSchedulerToken().length());
+
+			scheduleInfo = populateTaskDefinitionData(appName, scheduleInfo);
+
+			scheduleInfo.setCommandLineArgs(container.getArgs());
+			scheduleInfo.setScheduleProperties(new HashMap<>());
+			scheduleInfo.getScheduleProperties().put("spring.cloud.scheduler.cron.expression", cronjob.getSpec().getSchedule());
+			result.add(scheduleInfo);
+		}
+		Collections.sort(result);
+		return result;
+	}
+
+	private boolean isScheduleMigratable(String scheduleName) {
+		boolean result;
+		if (this.migrateProperties.getScheduleNamesToMigrate().size() > 0) {
+			result = this.migrateProperties.getScheduleNamesToMigrate().contains(scheduleName);
+		}
+		else {
+			result = true;
+		}
+		return result;
+	}
+
+	/**
+	 * Extracts the command line args from convertScheduleInfo that are tagged as properties
+	 * and inserts them in the {@link ConvertScheduleInfo} app and deployer properties.
+	 * Also adds the properties required for the launched task to interact with dataflow via the db.
+	 * @param scheduleInfo A {@link ConvertScheduleInfo} to be enriched.
+	 * @return the enriched {@link ConvertScheduleInfo}.
+	 */
+	@Override
+	public ConvertScheduleInfo enrichScheduleMetadata(ConvertScheduleInfo scheduleInfo) {
+		List<String> commandLineArgs = new ArrayList<>();
+		for (String arg : scheduleInfo.getCommandLineArgs()) {
+			if (arg.startsWith(this.appArgPrefix)) {
+				scheduleInfo.getAppProperties().put(
+						extractAppKey(arg, this.appArgPrefixLength),
+						extractValue(arg));
+			}
+			else if (arg.startsWith(this.deployerArgPrefix)) {
+				scheduleInfo.getDeployerProperties().put(
+						extractDeployerKey(arg, this.deployerArgPrefixLength),
+						extractValue(arg));
+			}
+			else if (arg.startsWith(this.commandArgPrefix)) {
+				commandLineArgs.add(arg.substring(this.commandArgPrefixLength));
+			}
+			else {
+				if (arg.startsWith("--spring.cloud.scheduler.task.launcher.taskName")) {
+					continue;
+				}
+				commandLineArgs.add(arg);
+			}
+		}
+		if (scheduleInfo.isCTR()) {
+			scheduleInfo.getAppProperties().put("graph", scheduleInfo.getCtrDSL());
+		}
+
+		if (!scheduleInfo.getAppProperties().containsKey("spring.cloud.task.name")) {
+			scheduleInfo.getAppProperties().put("spring.cloud.task.name", scheduleInfo.getTaskDefinitionName());
+		}
+
+		addDBInfoToScheduleInfo(scheduleInfo);
+
+		if (scheduleInfo.isCTR()) {
+			commandLineArgs.add("--dataflow-server-uri=" + this.migrateProperties.getDataflowUrl());
+		}
+		scheduleInfo.setCommandLineArgs(commandLineArgs);
+
+		return scheduleInfo;
+	}
+
+	/**
+	 * Creates a new schedule that launches the task and deletes the SchedulerTaskLauncher schedule.
+	 * @param scheduler the deployer scheduler to build the new schedule.
+	 * @param scheduleInfo the schedule info containing the existing schedule.
+	 */
+	@Override
+	public void migrateSchedule(Scheduler scheduler, ConvertScheduleInfo scheduleInfo) {
+		logger.info("Migrating " + scheduleInfo);
+		if(scheduleInfo.getTaskResource() == null) {
+			logger.info(String.format("The task definition name for schedule %s does not exist in Data Flow",
+					scheduleInfo.getScheduleName()));
+			return;
+		}
+		int scheduleTokenIndex = scheduleInfo.getScheduleName().indexOf(
+				this.migrateProperties.getSchedulerToken()) - 1;
+		String scheduleName = scheduleInfo.getScheduleName().substring(0, scheduleTokenIndex);
+		String appName = scheduleInfo.getScheduleName().substring(
+				scheduleInfo.getScheduleName().indexOf(
+						this.migrateProperties.getSchedulerToken()) + this.migrateProperties.getSchedulerToken().length());
+		AppDefinition appDefinition = new AppDefinition(appName, scheduleInfo.getAppProperties());
+		logger.info(String.format("Extracting schedule specific properties for schedule %s", scheduleInfo.getScheduleName()));
+		Map<String, String> schedulerProperties = extractAndQualifySchedulerProperties(scheduleInfo.getScheduleProperties());
+		ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, scheduleInfo.getDeployerProperties(), scheduleInfo.getCommandLineArgs(), scheduleName, scheduleInfo.getTaskResource());
+		logger.info(String.format("Staging Schedule %s", scheduleName));
+		scheduler.schedule(scheduleRequest);
+		logger.info(String.format("Unscheduling original %s", scheduleInfo.getScheduleName()));
+		scheduler.unschedule(scheduleInfo.getScheduleName());
+	}
+
+	protected Map<String, String> extractAndQualifySchedulerProperties(Map<String, String> scheduleInfoProperties) {
+		scheduleInfoProperties.put("spring.cloud.scheduler.kubernetes.taskServiceAccountName", this.migrateProperties.getTaskServiceAccountName());
+		//get properties from the spec
+		return scheduleInfoProperties;
+	}
+
+	@Override
+	public TaskDefinition findTaskDefinitionByName(String taskDefinitionName) {
+		return null;
+	}
+}

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -25,8 +25,6 @@ import java.util.List;
  * @author Glenn Renfro
  */
 public class MigrateProperties {
-	private String schedulerTaskLauncherUrl = "maven://org.springframework.cloud:spring-cloud-dataflow-scheduler-task-launcher:2.3.0.BUILD-SNAPSHOT";
-
 	/**
 	 * The token for the updated schedules.
 	 */
@@ -37,61 +35,47 @@ public class MigrateProperties {
 	 */
 	private String taskLauncherPrefix = "tasklauncher";
 
-	private String dataflowServerUri = "http://localhost:9393";
-
-	/**
-	 * The global Java Options required for the applications to be launched by the schedulerTaskLauncher.
-	 */
-	private String javaOptions;
-
-	/**
-	 * The global timeout to be assigned to applications to be launched by the schedulerTaskLauncher.
-	 */
-	private String healthCheckTimeout;
-
-	/**
-	 * The global api timeout to be assigned to applications to be launched by scheduleTaskLauncher.
-	 */
-	private Long apiTimeout;
-
-	/**
-	 * Timeout for status API operations in milliseconds to be assigned to applications to be launched by scheduleTaskLauncher
-	 */
-	private Long statusTimeout;
-
-	/**
-	 * If set, the global override the timeout allocated for staging an app launched by the schedulefTaskLauncher.
-	 */
-	private Long stagingTimeout;
-
-	/**
-	 * If set, the global override the timeout allocated for starting an app launched by scheduleTaskLauncher.
-	 */
-	private Long startupTimeout;
-
-	/**
-	 * If set, the global override for the maximum number of concurrently running tasks.
-	 */
-	private Integer maximumConcurrentTasks;
-
-	/**
-	 * The number of seconds to wait for a schedule to complete.
-	 * This excludes the time it takes to stage the application on Cloud Foundry.
-	 */
-	private int scheduleTimeoutInSeconds = 30;
-
 	/**
 	 * Comma delimited list of schedules to migrate.  If empty then all schedules will be migrated.
 	 */
 	private List<String> scheduleNamesToMigrate = new ArrayList<>();
 
-	public String getSchedulerTaskLauncherUrl() {
-		return schedulerTaskLauncherUrl;
-	}
 
-	public void setSchedulerTaskLauncherUrl(String schedulerTaskLauncherUrl) {
+	/**
-		this.schedulerTaskLauncherUrl = schedulerTaskLauncherUrl;
+	 * The registered application name for the composed task runner.
-	}
+	 * Update this if the dataflow server use has a different composed task runner app name than the default.
+	 */
+	private String composedTaskRunnerRegisteredAppName = "composed-task-runner";
+
+	/**
+	 * The user name of the database that the migrated task will use.
+	 */
+	private String dbUserName;
+
+	/**
+	 * The password of the database that the migrated task will use.
+	 */
+	private String dbPassword;
+
+	/**
+	 * The url to the database that the migrated task will use.
+	 */
+	private String dbUrl;
+
+	/**
+	 * The driver class name to use for the database that the migrated task will use.
+	 */
+	private String dbDriverClassName;
+
+	/**
+	 * The url of the Spring Cloud Data Flow Server that migrated composed task runners should execute task launch commands.
+	 */
+	private String dataflowUrl="http://localhost:9393";
+
+	/**
+	 * Establish the name of the service account for the schedule.
+	 */
+	private String taskServiceAccountName = "default";
 
 	public String getSchedulerToken() {
 		return schedulerToken;
@@ -109,78 +93,6 @@ public class MigrateProperties {
 		this.taskLauncherPrefix = taskLauncherPrefix;
 	}
 
-	public String getDataflowServerUri() {
-		return dataflowServerUri;
-	}
-
-	public void setDataflowServerUri(String dataflowServerUri) {
-		this.dataflowServerUri = dataflowServerUri;
-	}
-
-	public int getScheduleTimeoutInSeconds() {
-		return scheduleTimeoutInSeconds;
-	}
-
-	public void setScheduleTimeoutInSeconds(int scheduleTimeoutInSeconds) {
-		this.scheduleTimeoutInSeconds = scheduleTimeoutInSeconds;
-	}
-
-	public String getJavaOptions() {
-		return javaOptions;
-	}
-
-	public void setJavaOptions(String javaOptions) {
-		this.javaOptions = javaOptions;
-	}
-
-	public String getHealthCheckTimeout() {
-		return healthCheckTimeout;
-	}
-
-	public void setHealthCheckTimeout(String healthCheckTimeout) {
-		this.healthCheckTimeout = healthCheckTimeout;
-	}
-
-	public Long getApiTimeout() {
-		return apiTimeout;
-	}
-
-	public void setApiTimeout(Long apiTimeout) {
-		this.apiTimeout = apiTimeout;
-	}
-
-	public Long getStatusTimeout() {
-		return statusTimeout;
-	}
-
-	public void setStatusTimeout(Long statusTimeout) {
-		this.statusTimeout = statusTimeout;
-	}
-
-	public Long getStagingTimeout() {
-		return stagingTimeout;
-	}
-
-	public void setStagingTimeout(Long stagingTimeout) {
-		this.stagingTimeout = stagingTimeout;
-	}
-
-	public Long getStartupTimeout() {
-		return startupTimeout;
-	}
-
-	public void setStartupTimeout(Long startupTimeout) {
-		this.startupTimeout = startupTimeout;
-	}
-
-	public Integer getMaximumConcurrentTasks() {
-		return maximumConcurrentTasks;
-	}
-
-	public void setMaximumConcurrentTasks(Integer maximumConcurrentTasks) {
-		this.maximumConcurrentTasks = maximumConcurrentTasks;
-	}
-
 	public List<String> getScheduleNamesToMigrate() {
 		return scheduleNamesToMigrate;
 	}
@@ -188,4 +100,60 @@ public class MigrateProperties {
 	public void setScheduleNamesToMigrate(List<String> scheduleNamesToMigrate) {
 		this.scheduleNamesToMigrate = scheduleNamesToMigrate;
 	}
+
+	public String getDbUserName() {
+		return dbUserName;
+	}
+
+	public void setDbUserName(String dbUserName) {
+		this.dbUserName = dbUserName;
+	}
+
+	public String getDbPassword() {
+		return dbPassword;
+	}
+
+	public void setDbPassword(String dbPassword) {
+		this.dbPassword = dbPassword;
+	}
+
+	public String getDbUrl() {
+		return dbUrl;
+	}
+
+	public void setDbUrl(String dbUrl) {
+		this.dbUrl = dbUrl;
+	}
+
+	public String getDbDriverClassName() {
+		return dbDriverClassName;
+	}
+
+	public void setDbDriverClassName(String dbDriverClassName) {
+		this.dbDriverClassName = dbDriverClassName;
+	}
+
+	public String getComposedTaskRunnerRegisteredAppName() {
+		return composedTaskRunnerRegisteredAppName;
+	}
+
+	public void setComposedTaskRunnerRegisteredAppName(String composedTaskRunnerRegisteredAppName) {
+		this.composedTaskRunnerRegisteredAppName = composedTaskRunnerRegisteredAppName;
+	}
+
+	public String getDataflowUrl() {
+		return dataflowUrl;
+	}
+
+	public void setDataflowUrl(String dataflowUrl) {
+		this.dataflowUrl = dataflowUrl;
+	}
+
+	public String getTaskServiceAccountName() {
+		return taskServiceAccountName;
+	}
+
+	public void setTaskServiceAccountName(String taskServiceAccountName) {
+		this.taskServiceAccountName = taskServiceAccountName;
+	}
 }

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateScheduleService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ScheduleProcessedException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/SchedulerSkipPolicy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.

dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/TaskDefinitionRepository.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.

dataflow-migrate-schedules/src/main/resources/application.properties

@@ -1 +1,2 @@
 logging.level.org.springframework.cloud.task=debug
+spring.main.web-application-type=none

dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/AbstractApplications.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -18,7 +18,6 @@ package io.spring.migrateschedule;
 
 import org.cloudfoundry.doppler.LogMessage;
 import org.cloudfoundry.operations.applications.ApplicationEvent;
-import org.cloudfoundry.operations.applications.ApplicationManifest;
 import org.cloudfoundry.operations.applications.ApplicationSshEnabledRequest;
 import org.cloudfoundry.operations.applications.ApplicationSummary;
 import org.cloudfoundry.operations.applications.Applications;
@@ -27,7 +26,6 @@ import org.cloudfoundry.operations.applications.DeleteApplicationRequest;
 import org.cloudfoundry.operations.applications.DisableApplicationSshRequest;
 import org.cloudfoundry.operations.applications.EnableApplicationSshRequest;
 import org.cloudfoundry.operations.applications.GetApplicationEventsRequest;
-import org.cloudfoundry.operations.applications.GetApplicationManifestRequest;
 import org.cloudfoundry.operations.applications.ListApplicationTasksRequest;
 import org.cloudfoundry.operations.applications.LogsRequest;
 import org.cloudfoundry.operations.applications.PushApplicationManifestRequest;

dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/BatchIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -17,24 +17,20 @@
 package io.spring.migrateschedule;
 
 import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 
-import io.spring.migrateschedule.service.ConvertScheduleInfo;
 import io.spring.migrateschedule.configuration.BatchConfiguration;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
+import io.spring.migrateschedule.service.ConvertScheduleInfo;
 import io.spring.migrateschedule.service.MigrateScheduleService;
 import io.spring.migrateschedule.service.TaskDefinitionRepository;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 
-import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
-import org.springframework.batch.core.JobParameters;
-import org.springframework.batch.core.JobParametersBuilder;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.test.JobLauncherTestUtils;
 import org.springframework.batch.test.context.SpringBatchTest;
@@ -70,7 +66,7 @@ import static org.mockito.Mockito.when;
 @DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
 public class BatchIntegrationTests {
 
-	public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName";
+	public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-myapp";
 	public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName";
 	public static final String DEFAULT_APP_NAME = "defaultAppName";
 
@@ -86,6 +82,9 @@ public class BatchIntegrationTests {
 	@MockBean
 	private TaskDefinitionRepository taskDefinitionRepository;
 
+	@MockBean
+	private AppRegistrationRepository appRegistrationRepository;
+
 	@MockBean
 	private Scheduler scheduler;
 

dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFGetSchedulesTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -42,6 +42,7 @@ import io.pivotal.scheduler.v1.jobs.ListJobsRequest;
 import io.pivotal.scheduler.v1.jobs.ListJobsResponse;
 import io.pivotal.scheduler.v1.jobs.ScheduleJobRequest;
 import io.pivotal.scheduler.v1.jobs.ScheduleJobResponse;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
 import io.spring.migrateschedule.service.CFMigrateSchedulerService;
 import io.spring.migrateschedule.service.ConvertScheduleInfo;
 import io.spring.migrateschedule.service.MigrateProperties;
@@ -74,7 +75,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
 import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties;
-import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
+import org.springframework.cloud.deployer.spi.task.TaskLauncher;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -92,7 +93,8 @@ public class CFGetSchedulesTests {
 	private MigrateProperties migrateProperties;
 	private TaskDefinitionRepository taskDefinitionRepository;
 	private SchedulerClient schedulerClient;
-	private Scheduler scheduler;
+	private AppRegistrationRepository appRegistrationRepository;
+	private TaskLauncher taskLauncher;
 
 	@BeforeEach
 		public void setup() {
@@ -102,11 +104,14 @@ public class CFGetSchedulesTests {
 		this.cloudFoundryConnectionProperties.setSpace(DEFAULT_SPACE);
 		this.migrateProperties = new MigrateProperties();
 		this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class);
-		this.scheduler = Mockito.mock(Scheduler.class);
+		this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class);
+		this.taskLauncher = Mockito.mock(TaskLauncher.class);
+
 		this.cfConvertSchedulerService = new CFMigrateSchedulerService(this.cloudFoundryOperations,
 				this.schedulerClient,
 				this.cloudFoundryConnectionProperties, this.migrateProperties,
-				this.taskDefinitionRepository, new MavenProperties()) ;
+				this.taskDefinitionRepository, new MavenProperties(),
+				this.appRegistrationRepository, this.taskLauncher) ;
 	}
 
 

dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFMigrateScheduleConfigurationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2020 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.
@@ -17,12 +17,15 @@
 package io.spring.migrateschedule;
 
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.HashMap;
 
 import io.pivotal.scheduler.SchedulerClient;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
+import io.spring.migrateschedule.service.CFMigrateSchedulerService;
 import io.spring.migrateschedule.service.ConvertScheduleInfo;
 import io.spring.migrateschedule.service.MigrateProperties;
-import io.spring.migrateschedule.service.CFMigrateSchedulerService;
 import io.spring.migrateschedule.service.TaskDefinitionRepository;
 import org.cloudfoundry.operations.CloudFoundryOperations;
 import org.cloudfoundry.operations.applications.ApplicationDetail;
@@ -39,24 +42,27 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import reactor.core.publisher.Mono;
 
+import org.springframework.cloud.dataflow.core.AppRegistration;
 import org.springframework.cloud.dataflow.core.TaskDefinition;
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
 import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties;
 import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest;
 import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
+import org.springframework.cloud.deployer.spi.task.TaskLauncher;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 public class CFMigrateScheduleConfigurationTests {
 
-	public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName";
 	public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName";
+	public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-" + DEFAULT_TASK_DEFINITION_NAME;
 	public static final String DEFAULT_APP_NAME = "defaultAppName";
-	public static final String DEFAULT_CMD_ARG = "defaultCmd=WOW";
+	public static final String DEFAULT_CMD_ARG = "defaultCmd=MUCHWOW";
+	public static final String TASK_LAUNCHER_TASK_NAME_ARG = "--spring.cloud.scheduler.task.launcher.taskName="+ DEFAULT_TASK_DEFINITION_NAME;
 	public static final String DEFAULT_BUILD_PACK = "defaultBuildPack";
+	public static final String DEFAULT_DATA_FLOW_URI = "http://localhost:9393";
 
 	private CFMigrateSchedulerService cfConvertSchedulerService;
 	private CloudFoundryOperations cloudFoundryOperations;
@@ -65,6 +71,8 @@ public class CFMigrateScheduleConfigurationTests {
 	private TaskDefinitionRepository taskDefinitionRepository;
 	private SchedulerClient schedulerClient;
 	private Scheduler scheduler;
+	private AppRegistrationRepository appRegistrationRepository;
+	private TaskLauncher taskLauncher;
 
 	@BeforeEach
 		public void setup() {
@@ -74,10 +82,16 @@ public class CFMigrateScheduleConfigurationTests {
 		this.migrateProperties = new MigrateProperties();
 		this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class);
 		this.scheduler = Mockito.mock(Scheduler.class);
+		this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class);
+		this.taskLauncher = Mockito.mock(TaskLauncher.class);
+
+
 		this.cfConvertSchedulerService = new CFMigrateSchedulerService(this.cloudFoundryOperations,
 				this.schedulerClient,
 				this.cloudFoundryConnectionProperties, this.migrateProperties,
-				this.taskDefinitionRepository, new MavenProperties()) ;
+				this.taskDefinitionRepository, new MavenProperties(),
+				this.appRegistrationRepository,
+				this.taskLauncher) ;
 
 	}
 
@@ -85,15 +99,15 @@ public class CFMigrateScheduleConfigurationTests {
 	public void testEnrichment() {
 		ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo();
 		assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(5);
-		assertThat(scheduleInfo.getAppProperties().get("tasklauncher.app.defaultAppName.foo")).isEqualTo("bar");
+		assertThat(scheduleInfo.getAppProperties().get("foo")).isEqualTo("bar");
-		assertThat(scheduleInfo.getAppProperties().get("spring.cloud.dataflow.client.serverUri")).isEqualTo("http://localhost:9393");
+		assertThat(scheduleInfo.getAppProperties().get("dataflow-server-uri")).isEqualTo(DEFAULT_DATA_FLOW_URI);
-		assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.memory")).isEqualTo("1024m");
+		assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.memory")).isEqualTo("1024m");
-		assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.health-check")).isEqualTo("port");
+		assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.health-check")).isEqualTo("port");
-		assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.disk")).isEqualTo("1024m");
+		assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.disk")).isEqualTo("1024m");
 
 		assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(2);
-		assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("cmdarg.tasklauncher.defaultCmd=WOW");
+		assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo(DEFAULT_CMD_ARG);
-		assertThat(scheduleInfo.getCommandLineArgs().get(1)).isEqualTo("--spring.cloud.scheduler.task.launcher.taskName=defaultTaskDefinitionName");
+		assertThat(scheduleInfo.getCommandLineArgs().get(1)).isEqualTo("--spring.application.name=defaultTaskDefinitionName");
 	}
 
 	@Test
@@ -103,18 +117,21 @@ public class CFMigrateScheduleConfigurationTests {
 		scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME);
 		scheduleInfo.setScheduleProperties(new HashMap<>());
 		scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME);
+		scheduleInfo.getCommandLineArgs().add(TASK_LAUNCHER_TASK_NAME_ARG);
 		Mockito.when(cloudFoundryOperations.applications()).thenReturn(new NoPropertyApplication());
+		createMockAppRegistration();
 		TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder
 				.from(new TaskDefinition("fooTask", "foo"))
 				.setTaskName(DEFAULT_TASK_DEFINITION_NAME)
 				.setRegisteredAppName(DEFAULT_APP_NAME)
+				.setDslText("timestamp")
 				.build();
 		Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition);
 		scheduleInfo = this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo);
 		assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(1);
-		assertThat(scheduleInfo.getAppProperties().get("spring.cloud.dataflow.client.serverUri")).isEqualTo("http://localhost:9393");
+		assertThat(scheduleInfo.getAppProperties().get("dataflow-server-uri")).isEqualTo("http://localhost:9393");
 		assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(1);
-		assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("--spring.cloud.scheduler.task.launcher.taskName=defaultTaskDefinitionName");
+		assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("--spring.application.name=defaultTaskDefinitionName");
 	}
 
 	@Test
@@ -126,10 +143,10 @@ public class CFMigrateScheduleConfigurationTests {
 		scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME);
 		scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG);
 		Mockito.when(cloudFoundryOperations.applications()).thenReturn(new SinglePropertyApplication());
-		assertThrows(IllegalStateException.class, () -> this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo));
+		scheduleInfo = this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo);
+		assertThat(scheduleInfo.getTaskResource()).isNull();
 	}
 
-
 	@Test
 	public void testMigrate() {
 		ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo();
@@ -137,8 +154,8 @@ public class CFMigrateScheduleConfigurationTests {
 		final ArgumentCaptor<ScheduleRequest> scheduleRequestArgument = ArgumentCaptor.forClass(ScheduleRequest.class);
 		final ArgumentCaptor<String> scheduleNameArg = ArgumentCaptor.forClass(String.class);
 		verify(this.scheduler, times(1)).schedule(scheduleRequestArgument.capture());
-		verify(this.scheduler, times(1)).unschedule(scheduleNameArg.capture());
+		verify(this.taskLauncher, times(1)).destroy(scheduleNameArg.capture());
-		assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName-scdf-defaultTaskDefinitionName");
+		assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName");
 		assertThat(scheduleNameArg.getValue()).isEqualTo(DEFAULT_SCHEDULE_NAME);
 	}
 
@@ -149,13 +166,17 @@ public class CFMigrateScheduleConfigurationTests {
 		scheduleInfo.setScheduleProperties(new HashMap<>());
 		scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME);
 		scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG);
+		scheduleInfo.getCommandLineArgs().add(TASK_LAUNCHER_TASK_NAME_ARG);
+
 		Mockito.when(cloudFoundryOperations.applications()).thenReturn(new SinglePropertyApplication());
 		TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder
 				.from(new TaskDefinition("fooTask", "foo"))
 				.setTaskName(DEFAULT_TASK_DEFINITION_NAME)
 				.setRegisteredAppName(DEFAULT_APP_NAME)
+				.setDslText("timestamp")
 				.build();
 		Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition);
+		createMockAppRegistration();
 		return this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo);
 	}
 
@@ -217,4 +238,15 @@ public class CFMigrateScheduleConfigurationTests {
 			return Mono.empty();
 		}
 	}
+
+	private void createMockAppRegistration() {
+		AppRegistration appRegistration = new AppRegistration();
+		try {
+			appRegistration.setUri(new URI("docker://fun:party"));
+		}
+		catch (URISyntaxException e) {
+			e.printStackTrace();
+		}
+		Mockito.when(this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(Mockito.any(), Mockito.any())).thenReturn(appRegistration);
+	}
 }

dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/KubernetesMigrateScheduleConfigurationTests.java

@@ -0,0 +1,219 @@
+/*
+ * Copyright 2020 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 io.spring.migrateschedule;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.PodSpec;
+import io.fabric8.kubernetes.api.model.PodTemplateSpec;
+import io.fabric8.kubernetes.api.model.batch.CronJob;
+import io.fabric8.kubernetes.api.model.batch.CronJobList;
+import io.fabric8.kubernetes.api.model.batch.CronJobSpec;
+import io.fabric8.kubernetes.api.model.batch.DoneableCronJob;
+import io.fabric8.kubernetes.api.model.batch.JobSpec;
+import io.fabric8.kubernetes.api.model.batch.JobTemplateSpec;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.BatchAPIGroupDSL;
+import io.fabric8.kubernetes.client.dsl.MixedOperation;
+import io.spring.migrateschedule.service.AppRegistrationRepository;
+import io.spring.migrateschedule.service.ConvertScheduleInfo;
+import io.spring.migrateschedule.service.KubernetesMigrateSchedulerService;
+import io.spring.migrateschedule.service.MigrateProperties;
+import io.spring.migrateschedule.service.TaskDefinitionRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import org.springframework.cloud.dataflow.core.AppRegistration;
+import org.springframework.cloud.dataflow.core.TaskDefinition;
+import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest;
+import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
+import org.springframework.core.io.Resource;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class KubernetesMigrateScheduleConfigurationTests {
+
+	public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-mydef";
+	public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName";
+	public static final String DEFAULT_APP_NAME = "defaultAppName";
+	public static final String DEFAULT_CMD_ARG = "defaultCmd=WOW";
+	public static final String DEFAULT_BUILD_PACK = "defaultBuildPack";
+	public static final String DEFAULT_DB_URL = "jdbc:mydb://localhost:8888/task";
+	public static final String DEFAULT_DRIVER_CLASS_NAME = "myDriverClassName";
+	public static final String DEFAULT_DB_USER_NAME = "myuser";
+	public static final String DEFAULT_DB_PASSWORD = "mypassword";
+
+	private KubernetesMigrateSchedulerService k8MigrateSchedulerService;
+	private KubernetesClient kubernetesClient;
+	private MigrateProperties migrateProperties;
+	private TaskDefinitionRepository taskDefinitionRepository;
+	private Scheduler scheduler;
+	private AppRegistrationRepository appRegistrationRepository;
+	private List<CronJob> sampleCronJobs;
+
+	@BeforeEach
+		public void setup() {
+		this.kubernetesClient = Mockito.mock(KubernetesClient.class);
+		this.migrateProperties = new MigrateProperties();
+		this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class);
+		this.scheduler = Mockito.mock(Scheduler.class);
+		this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class);
+		this.migrateProperties.setDbUrl(DEFAULT_DB_URL);
+		this.migrateProperties.setDbDriverClassName(DEFAULT_DRIVER_CLASS_NAME);
+		this.migrateProperties.setDbUserName(DEFAULT_DB_USER_NAME);
+		this.migrateProperties.setDbPassword(DEFAULT_DB_PASSWORD);
+
+		this.k8MigrateSchedulerService = new KubernetesMigrateSchedulerService(this.taskDefinitionRepository,
+				this.appRegistrationRepository,
+				this.kubernetesClient, this.migrateProperties) ;
+
+		BatchAPIGroupDSL batchAPIGroupDSL = Mockito.mock(BatchAPIGroupDSL.class);
+		MixedOperation<CronJob, CronJobList, DoneableCronJob, io.fabric8.kubernetes.client.dsl.Resource<CronJob, DoneableCronJob>> cronJobs = Mockito.mock(MixedOperation.class);
+		CronJobList cronJobList = Mockito.mock(CronJobList.class);
+		Mockito.when(this.kubernetesClient.batch()).thenReturn(batchAPIGroupDSL);
+		Mockito.when(batchAPIGroupDSL.cronjobs()).thenReturn(cronJobs);
+		Mockito.when(cronJobs.list()).thenReturn(cronJobList);
+		sampleCronJobs = new ArrayList<>();
+		Mockito.when(cronJobList.getItems()).thenReturn(sampleCronJobs);
+
+		//Setup kubernetesClient Mock to return a single schedule item in the result
+		CronJob cronJob = new CronJob();
+		ObjectMeta objectMeta = new ObjectMeta();
+		objectMeta.setName(DEFAULT_SCHEDULE_NAME);
+		cronJob.setMetadata(objectMeta);
+		this.sampleCronJobs.add(cronJob);
+		CronJobSpec cronJobSpec = new CronJobSpec();
+		cronJob.setSpec(cronJobSpec);
+		JobTemplateSpec jobTemplateSpec = new JobTemplateSpec();
+		cronJobSpec.setJobTemplate(jobTemplateSpec);
+		JobSpec jobSpec = new JobSpec();
+		jobTemplateSpec.setSpec(jobSpec);
+		PodTemplateSpec podTemplateSpec = new PodTemplateSpec();
+		jobSpec.setTemplate(podTemplateSpec);
+		PodSpec podSpec = new PodSpec();
+		podTemplateSpec.setSpec(podSpec);
+
+		Container container = new Container();
+		podSpec.getContainers().add(container);
+
+	}
+
+	@Test
+	public void testEnrichment() {
+		ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo();
+		assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(6);
+		assertThat(scheduleInfo.getAppProperties().get("foo")).isEqualTo("bar");
+		assertThat(scheduleInfo.getDeployerProperties().get("que")).isEqualTo("qix");
+
+		validateDBProperties(scheduleInfo);
+		assertThat(scheduleInfo.getAppProperties().get("spring.cloud.task.name")).isEqualTo(DEFAULT_TASK_DEFINITION_NAME);
+
+		assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(1);
+		assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("defaultCmd=WOW");
+	}
+
+	@Test
+	public void testEnrichmentNoProps() {
+		ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo();
+		scheduleInfo.setScheduleName(DEFAULT_SCHEDULE_NAME);
+		scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME);
+		scheduleInfo.setScheduleProperties(new HashMap<>());
+		scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME);
+		TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder
+				.from(new TaskDefinition("fooTask", "foo"))
+				.setTaskName(DEFAULT_TASK_DEFINITION_NAME)
+				.setRegisteredAppName(DEFAULT_APP_NAME)
+				.build();
+		Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition);
+		scheduleInfo = this.k8MigrateSchedulerService.enrichScheduleMetadata(scheduleInfo);
+		assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(5);
+		validateDBProperties(scheduleInfo);
+		assertThat(scheduleInfo.getAppProperties().get("spring.cloud.task.name")).isEqualTo(DEFAULT_TASK_DEFINITION_NAME);
+		assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(0);
+	}
+
+	@Test
+	public void testListSchedules() {
+
+		TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder
+				.from(new TaskDefinition("fooTask", "foo"))
+				.setTaskName(DEFAULT_TASK_DEFINITION_NAME)
+				.setRegisteredAppName(DEFAULT_APP_NAME)
+				.setDslText("foo")
+				.build();
+		AppRegistration appRegistration = new AppRegistration();
+		try {
+			appRegistration.setUri(new URI("docker://fun:party"));
+		}
+		catch (URISyntaxException e) {
+			e.printStackTrace();
+		}
+		Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition);
+		Mockito.when(this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(Mockito.any(), Mockito.any())).thenReturn(appRegistration);
+		List<ConvertScheduleInfo>  convertScheduleInfos = this.k8MigrateSchedulerService.scheduleInfoList();
+		assertThat(convertScheduleInfos.size()).isEqualTo(1);
+		assertThat(convertScheduleInfos.get(0).getScheduleName()).isEqualTo(DEFAULT_SCHEDULE_NAME);
+	}
+
+	@Test
+	public void testMigrate() {
+		ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo();
+		this.k8MigrateSchedulerService.migrateSchedule(this.scheduler, scheduleInfo);
+		final ArgumentCaptor<ScheduleRequest> scheduleRequestArgument = ArgumentCaptor.forClass(ScheduleRequest.class);
+		final ArgumentCaptor<String> scheduleNameArg = ArgumentCaptor.forClass(String.class);
+		verify(this.scheduler, times(1)).schedule(scheduleRequestArgument.capture());
+		verify(this.scheduler, times(1)).unschedule(scheduleNameArg.capture());
+		assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName");
+		assertThat(scheduleNameArg.getValue()).isEqualTo(DEFAULT_SCHEDULE_NAME);
+	}
+
+	private ConvertScheduleInfo createFoundationConvertScheduleInfo() {
+		ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo();
+		scheduleInfo.setScheduleName(DEFAULT_SCHEDULE_NAME);
+		scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME);
+		scheduleInfo.setScheduleProperties(new HashMap<>());
+		scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME);
+		scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG);
+		scheduleInfo.setTaskResource(Mockito.mock(Resource.class));
+		TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder
+				.from(new TaskDefinition("fooTask", "foo"))
+				.setTaskName(DEFAULT_TASK_DEFINITION_NAME)
+				.setRegisteredAppName(DEFAULT_APP_NAME)
+				.build();
+		Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition);
+		scheduleInfo.getAppProperties().put("foo", "bar");
+		scheduleInfo.getDeployerProperties().put("que", "qix");
+		return this.k8MigrateSchedulerService.enrichScheduleMetadata(scheduleInfo);
+	}
+
+	private void validateDBProperties(ConvertScheduleInfo scheduleInfo) {
+		assertThat(scheduleInfo.getAppProperties().get("spring.datasource.url")).isEqualTo(DEFAULT_DB_URL);
+		assertThat(scheduleInfo.getAppProperties().get("spring.datasource.username")).isEqualTo(DEFAULT_DB_USER_NAME);
+		assertThat(scheduleInfo.getAppProperties().get("spring.datasource.password")).isEqualTo(DEFAULT_DB_PASSWORD);
+		assertThat(scheduleInfo.getAppProperties().get("spring.datasource.driverClassName")).isEqualTo(DEFAULT_DRIVER_CLASS_NAME);
+	}
+}