Implement Tasklauncher function

Clean up test

applications/sink/pom.xml

@@ -23,6 +23,7 @@
         <module>rabbit-sink</module>
         <module>router-sink</module>
         <module>sftp-sink</module>
+        <module>tasklauncher-sink</module>
         <module>tcp-sink</module>
         <module>throughput-sink</module>
         <module>websocket-sink</module>

applications/sink/tasklauncher-sink/README.adoc

@@ -0,0 +1,99 @@
+//tag::ref-doc[]
+= Tasklauncher Sink
+
+This module consumes LaunchRequest messages from a `PollableMessageSource` and uses the Data Flow REST client to launch a registered task on a configured https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-tasks[task platform].
+The client must be configured to connect to a remote Data Flow Server, including any required authentication (see Configuration Options below).
+
+This application launches a registered task definition using the Data Flow Server https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#api-guide-resources-task-executions-launching[REST API].
+
+== Input
+
+a Task Launch Request including:
+
+* the task name (required and created as a task with the target Data Flow Server)
+* deployment properties (key value pairs, optional).
+* program arguments for the task (a list, optional).
+
+The message payload can be a JSON document:
+
+[source,json]
+----
+{
+  "name":"foo",
+  "deploymentProps": {"key1":"val1","key2":"val2"},
+  "args":["--debug", "--foo", "bar"]
+}
+----
+
+Minimally, it must include the task name.
+
+[source,json]
+----
+{"name":"foo"}
+----
+
+=== Options
+
+The **$$tasklauncher-dataflow$$** $$sink$$ supports the following configuration properties:
+
+//tag::configuration-properties[]
+$$spring.cloud.dataflow.client.authentication.access-token$$:: $$OAuth2 Access Token.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.basic.password$$:: $$The login password.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.basic.username$$:: $$The login username.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.client-id$$:: $$OAuth2 Client Id.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.client-secret$$:: $$OAuth2 Client Secret.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.oauth2.client-registration-id$$:: $$<documentation missing>$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.oauth2.password$$:: $$<documentation missing>$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.oauth2.username$$:: $$<documentation missing>$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.scope$$:: $$OAuth2 Scopes.$$ *($$Set<String>$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.authentication.token-uri$$:: $$OAuth2 Token Uri.$$ *($$String$$, default: `$$<none>$$`)*
+$$spring.cloud.dataflow.client.enable-dsl$$:: $$Enable Data Flow DSL access.$$ *($$Boolean$$, default: `$$false$$`)*
+$$spring.cloud.dataflow.client.server-uri$$:: $$The Data Flow server URI.$$ *($$String$$, default: `$$http://localhost:9393$$`)*
+$$spring.cloud.dataflow.client.skip-ssl-validation$$:: $$Skip Ssl validation.$$ *($$Boolean$$, default: `$$false$$`)*
+$$tasklauncher.platform-name$$:: $$The Spring Cloud Data Flow platform to use for launching tasks.$$ *($$String$$, default: `$$default$$`)*
+$$trigger.initial-delay$$:: $$The initial delay in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)*
+$$trigger.max-period$$:: $$The maximum polling period in milliseconds. Will be set to period if period > maxPeriod.$$ *($$Integer$$, default: `$$30000$$`)*
+$$trigger.period$$:: $$The polling period in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)*
+//end::configuration-properties[]
+
+== Using the TaskLauncher
+The tasklauncher sink consumes `LaunchRequest` messages, as described above, and launches a task using the target Data Flow server (given by `--spring.cloud.dataflow.client.server-uri`).
+The task launcher periodically polls its input source for launch requests but will pause polling when the platform has reached it's concurrent task execution limit, given by `spring.cloud.dataflow.task.platform.<platform-type>.accounts[<account-name>].maximum-concurrent-tasks`.
+This prevents the SCDF deployer's deployment platform from exhausting its resources under heavy task load.
+The poller is scheduled using a `DynamicPeriodicTrigger`. By default the initial polling rate is 1 second, but may be configured to any duration. When polling is paused, or if there are no launch requests present, the trigger period will increase, applying exponential backoff, up to a configured maximum (30 seconds by default).
+
+NOTE: This version of the Data Flow task launcher requires SCDF version 2.4.x or higher
+
+The SCDF server may be configured to launch tasks on multiple platforms.
+Each task launcher instance is configured for a single platform, given by the `platformName` property (`default` if not specified).
+This limitation is enforced because if the server has multiple task platforms configured, it may be the case that some of its task platforms are at the limit and some are not.
+In this situation, we can only consume the next launch request if we know for which task platform it is targeted.
+For this reason, if the SCDF server is configured for multiple task platforms (or a single non-default platform), we assume that all launch requests are targeted for that platform.
+The task launcher will set the required deployment property `spring.cloud.dataflow.task.platformName` if the request does not provide it.
+
+NOTE: If the request includes the deployment property `spring.cloud.dataflow.task.platformName`, and the value is not the same as the tasklauncher's `platformName`, the task launcher will throw an exception.
+
+To launch tasks on multiple platforms, you must configure a task launcher instance per platform and use a link:../router-sink/[router sink], or https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#partitioning[partitioning strategy], to route requests to the correct instance.
+
+NOTE: When the poller is paused it puts pressure
+on the message broker so some tuning will be necessary in extreme cases to balance resource utilization.
+
+=== Client Authentication
+
+If the Data Flow server requires authentication, the client must pass credentials with authorization to launch a task.
+The Data Flow client supports both basic and OAuth2 authentication.
+
+For basic authentication set the username and password:
+
+```
+--spring.cloud.dataflow.client.authentication.basic.username=<username> --spring.cloud.dataflow.client.authentication.basic.password=<password>
+```
+
+For OAuth2 authentication, set the `client-id`, `client-secret`, and `token-uri` at a minimum. These values correspond to values set in the SCDF server's OAuth2 configuration.
+For more details, see https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-security[the Security section in the Data Flow reference].
+
+```
+--spring.cloud.dataflow.client.authentication.client-id=<client-id> --spring.cloud.dataflow.client.authentication.client-secret=<client-secret> spring.cloud.dataflow.client.authentication.token-uri: <token-uri>
+```
+
+//end::ref-doc[]

applications/sink/tasklauncher-sink/pom.xml

@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>tasklauncher-sink</artifactId>
+    <version>3.0.0-SNAPSHOT</version>
+    <name>tasklauncher-sink</name>
+    <description>tasklauncher sink app</description>
+    <packaging>jar</packaging>
+
+    <parent>
+        <groupId>org.springframework.cloud.stream.app</groupId>
+        <artifactId>stream-applications-core</artifactId>
+        <version>3.0.0-SNAPSHOT</version>
+        <relativePath/>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>tasklauncher-function</artifactId>
+            <version>${java-functions.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.springframework.cloud.stream.app.plugin</groupId>
+                <artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
+                <configuration>
+                    <generatedApp>
+                        <name>tasklauncher</name>
+                        <type>sink</type>
+                        <version>${project.version}</version>
+                        <configClass>org.springframework.cloud.stream.app.tasklauncher.sink.TaskLauncherSinkConfiguration.class</configClass>
+                    </generatedApp>
+                    <dependencies>
+                        <dependency>
+                            <groupId>org.springframework.cloud.stream.app</groupId>
+                            <artifactId>tasklauncher-sink</artifactId>
+                            <version>${project.version}</version>
+                        </dependency>
+                    </dependencies>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <repositories>
+        <repository>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+            <id>spring-snapshots</id>
+            <name>Spring Snapshots</name>
+            <url>https://repo.spring.io/libs-snapshot-local</url>
+        </repository>
+        <repository>
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo.spring.io/libs-milestone-local</url>
+        </repository>
+    </repositories>
+
+</project>

applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/LaunchRequestConsumer.java

@@ -0,0 +1,210 @@
+/*
+ * Copyright 2020-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.tasklauncher.sink;
+
+import java.time.Duration;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.fn.tasklauncher.LaunchRequest;
+import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunction;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.integration.util.DynamicPeriodicTrigger;
+import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
+import org.springframework.util.Assert;
+
+/**
+ *
+ * A Message consumer that submits received task {@link LaunchRequest}s to a Data Flow
+ * server. This polls a {@link PollableMessageSource} only if the Data Flow server is not
+ * at its concurrent task execution limit.
+ *
+ * The consumer runs as a {@link ScheduledFuture} , configured with a
+ * {@link DynamicPeriodicTrigger} to support exponential backoff up to a maximum period.
+ * Every period cycle, the poller first makes a REST call to the Data Flow server to check
+ * if it can accept a new task LaunchRequest before checking the Message source. The
+ * polling period will back off (increase) when either the server is not accepting
+ * requests or no request is received.
+ *
+ * The period will revert to its initial value whenever both a request is received and the
+ * DataFlow Server is accepting launch requests. The period remain at the maximum value
+ * when there are no requests to avoid hammering the Data Flow server for no reason.
+ *
+ * @author David Turanski
+ **/
+public class LaunchRequestConsumer implements SmartLifecycle {
+	private static final Log log = LogFactory.getLog(LaunchRequestConsumer.class);
+
+	private static final int BACKOFF_MULTIPLE = 2;
+
+	static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName";
+
+	private final PollableMessageSource input;
+
+	private final AtomicBoolean running = new AtomicBoolean();
+
+	private final AtomicBoolean paused = new AtomicBoolean();
+
+	private final DynamicPeriodicTrigger trigger;
+
+	private final ConcurrentTaskScheduler taskScheduler;
+
+	private final long initialPeriod;
+
+	private final long maxPeriod;
+
+	private volatile boolean autoStart = true;
+
+	private final TaskLauncherFunction taskLauncherFunction;
+
+	private ScheduledFuture<?> scheduledFuture;
+
+	public LaunchRequestConsumer(PollableMessageSource input, DynamicPeriodicTrigger trigger,
+			long maxPeriod, TaskLauncherFunction taskLauncherFunction) {
+		Assert.notNull(input, "`input` cannot be null.");
+		Assert.notNull(taskLauncherFunction, "`taskLauncherFunction` cannot be null.");
+		this.taskLauncherFunction = taskLauncherFunction;
+		this.input = input;
+		this.trigger = trigger;
+		this.initialPeriod = trigger.getDuration().toMillis();
+		this.maxPeriod = maxPeriod;
+		this.taskScheduler = new ConcurrentTaskScheduler();
+	}
+
+	/*
+	 * Polling loop
+	 */
+	ScheduledFuture<?> consume() {
+
+		return taskScheduler.schedule(() -> {
+			if (!isRunning()) {
+				return;
+			}
+
+			if (taskLauncherFunction.platformIsAcceptingNewTasks()) {
+				if (paused.compareAndSet(true, false)) {
+					log.info("Polling resumed");
+				}
+
+				if (!input.poll(message -> {
+					LaunchRequest request = (LaunchRequest) message.getPayload();
+					log.debug("Received a Task launch request - task name:  " + request.getTaskName());
+					taskLauncherFunction.apply(request);
+				}, new ParameterizedTypeReference<LaunchRequest>() {
+				})) {
+					backoff("No task launch request received");
+				}
+				else {
+					if (trigger.getDuration().toMillis() > initialPeriod) {
+						trigger.setDuration(Duration.ofMillis(initialPeriod));
+						log.info(String.format("Polling period reset to %d ms.", trigger.getDuration().toMillis()));
+					}
+				}
+			}
+			else {
+				paused.set(true);
+				backoff("Polling paused");
+
+			}
+		}, trigger);
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return autoStart;
+	}
+
+	public void setAutoStartup(boolean autoStart) {
+		this.autoStart = autoStart;
+	}
+
+	@Override
+	public synchronized void stop(Runnable callback) {
+		if (callback != null) {
+			callback.run();
+		}
+		this.stop();
+	}
+
+	@Override
+	public void start() {
+		if (running.compareAndSet(false, true)) {
+			this.scheduledFuture = consume();
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (running.getAndSet(false)) {
+			this.scheduledFuture.cancel(false);
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return running.get();
+	}
+
+	public boolean isPaused() {
+		return paused.get();
+	}
+
+	@Override
+	public int getPhase() {
+		return Integer.MAX_VALUE;
+	}
+
+	private void backoff(String message) {
+		synchronized (trigger) {
+			if (trigger.getDuration().compareTo(Duration.ZERO) > 0
+					&& trigger.getDuration().compareTo(Duration.ofMillis(maxPeriod)) < 0) {
+
+				Duration duration = trigger.getDuration();
+
+				if (duration.multipliedBy(BACKOFF_MULTIPLE).compareTo(Duration.ofMillis(maxPeriod)) <= 0) {
+					// If d >= 1, round to 1 seconds.
+					if (duration.getSeconds() == 1) {
+						duration = Duration.ofSeconds(1);
+					}
+					duration = duration.multipliedBy(BACKOFF_MULTIPLE);
+				}
+				else {
+					duration = Duration.ofMillis(maxPeriod);
+				}
+				if (trigger.getDuration().toMillis() < 1000) {
+					log.info(String.format(message + " - increasing polling period to %d ms.", duration.toMillis()));
+				}
+				else {
+					log.info(
+							String.format(message + "- increasing polling period to %d seconds.",
+									duration.getSeconds()));
+				}
+
+				trigger.setDuration(duration);
+			}
+			else if (trigger.getDuration() == Duration.ofMillis(maxPeriod)) {
+				log.info(message);
+			}
+		}
+	}
+
+}

applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/PollingSink.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2018-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.tasklauncher.sink;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+
+/**
+ * @author David Turanski
+ **/
+public interface PollingSink {
+	/**
+	 * The input name.
+	 */
+	String INPUT = "input";
+
+	@Input(PollingSink.INPUT)
+	PollableMessageSource input();
+}

applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkConfiguration.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2018-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.tasklauncher.sink;
+
+import java.time.Duration;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunction;
+import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunctionConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.util.DynamicPeriodicTrigger;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Configuration class for the TaskLauncher Data Flow Sink.
+ *
+ * @author David Turanski
+ * @author Gunnar Hillert
+ */
+@EnableBinding(PollingSink.class)
+@EnableConfigurationProperties({ TriggerProperties.class })
+@Import(TaskLauncherFunctionConfiguration.class)
+public class TaskLauncherSinkConfiguration {
+
+	@Value("${autostart:true}")
+	private boolean autoStart;
+
+	@Bean
+	public DynamicPeriodicTrigger periodicTrigger(TriggerProperties triggerProperties) {
+		DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(triggerProperties.getPeriod());
+		trigger.setInitialDuration(Duration.ofMillis(triggerProperties.getInitialDelay()));
+		return trigger;
+	}
+
+	/*
+	 * For backward compatibility with spring-cloud-stream-2.1.x
+	 */
+	@Bean
+	public BeanPostProcessor addInterceptorToPollableMessageSource() {
+		return new BeanPostProcessor() {
+			@Override
+			public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+				if (bean instanceof DefaultPollableMessageSource) {
+					DefaultPollableMessageSource pollableMessageSource = (DefaultPollableMessageSource) bean;
+					pollableMessageSource.addInterceptor(new ChannelInterceptor() {
+						@Override
+						public Message<?> preSend(Message<?> message, MessageChannel channel) {
+							Message<?> newMessage = message;
+							if (message.getHeaders().containsKey("originalContentType")) {
+								newMessage = MessageBuilder.fromMessage(message)
+										.setHeader(MessageHeaders.CONTENT_TYPE,
+												message.getHeaders().get("originalContentType"))
+										.build();
+							}
+							return newMessage;
+						}
+					});
+				}
+				return bean;
+			}
+		};
+	}
+
+	@Bean
+	public LaunchRequestConsumer launchRequestConsumer(PollableMessageSource input,
+			TaskLauncherFunction taskLauncherFunction, DynamicPeriodicTrigger trigger,
+			TriggerProperties triggerProperties) {
+
+		LaunchRequestConsumer consumer = new LaunchRequestConsumer(input,
+				trigger, triggerProperties.getMaxPeriod(), taskLauncherFunction);
+		consumer.setAutoStartup(autoStart);
+		return consumer;
+	}
+}

applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TriggerProperties.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2018-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.tasklauncher.sink;
+
+import javax.annotation.PostConstruct;
+import javax.validation.constraints.Min;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * @author David Turanski
+ **/
+@ConfigurationProperties(prefix = "trigger")
+@Validated
+public class TriggerProperties {
+	/**
+	 * The initial delay in milliseconds.
+	 */
+	private int initialDelay = 1000;
+
+	/**
+	 * The polling period in milliseconds.
+	 */
+	private int period = 1000;
+
+	/**
+	 * The maximum polling period in milliseconds. Will be set to period if period >
+	 * maxPeriod.
+	 */
+	private int maxPeriod = 30000;
+
+	@Min(0)
+	public int getInitialDelay() {
+		return initialDelay;
+	}
+
+	public void setInitialDelay(int initialDelay) {
+		this.initialDelay = initialDelay;
+	}
+
+	@Min(0)
+	public int getPeriod() {
+		return period;
+	}
+
+	public void setPeriod(int period) {
+		this.period = period;
+	}
+
+	@Min(1000)
+	public int getMaxPeriod() {
+		return maxPeriod;
+	}
+
+	public void setMaxPeriod(int maxPeriod) {
+		this.maxPeriod = maxPeriod;
+	}
+
+	@PostConstruct
+	public void checkMaxPeriod() {
+		maxPeriod = Integer.max(maxPeriod, period);
+	}
+}

applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties

@@ -0,0 +1,9 @@
+configuration-properties.classes=org.springframework.cloud.fn.tasklauncher.TaskLauncherFunctionProperties,\
+  org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties,\
+  org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication,\
+  org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Basic,\
+   org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Oauth2,\
+  org.springframework.cloud.stream.app.tasklauncher.sink.TriggerProperties
+
+
+

applications/sink/tasklauncher-sink/src/test/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkTests.java

@@ -0,0 +1,312 @@
+/*
+ * Copyright 2020-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.tasklauncher.sink;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.stubbing.Answer;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.dataflow.rest.client.DataFlowOperations;
+import org.springframework.cloud.dataflow.rest.client.TaskOperations;
+import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource;
+import org.springframework.cloud.dataflow.rest.resource.LauncherResource;
+import org.springframework.cloud.fn.tasklauncher.LaunchRequest;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.env.Environment;
+import org.springframework.hateoas.PagedModel;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.AcknowledgmentCallback;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.util.DynamicPeriodicTrigger;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author David Turanski
+ **/
+public class TaskLauncherSinkTests {
+
+	private ApplicationContextRunner contextRunner;
+
+	@BeforeEach
+	public void setUp() {
+		contextRunner = TestChannelBinderConfiguration.applicationContextRunner()
+				.withUserConfiguration(TaskLauncherSinkConfiguration.class, TestConfig.class);
+	}
+
+	@Test
+	public void consumerPausesWhenMaxTaskExecutionsReached() {
+		contextRunner = contextRunner.withPropertyValues(
+				"trigger.period=10",
+				"trigger.initial-delay=0",
+				"autostart=false")
+				.run(context -> {
+					CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource(
+							context);
+
+					LaunchRequestConsumer consumer = consumer(context);
+					CountDownLatch countDownLatch = countDownLatch(context);
+					DynamicPeriodicTrigger trigger = trigger(context);
+
+					consumer.start();
+
+					assertThat(countDownLatch.await(1, TimeUnit.SECONDS)).isTrue();
+					assertThat(currentTaskExecutionsResource.getRunningExecutionCount()).isEqualTo(
+							currentTaskExecutionsResource.getMaximumTaskExecutions());
+					assertThat(eventually(c -> c.isPaused() && c.isRunning(), consumer)).isTrue();
+
+					currentTaskExecutionsResource.setRunningExecutionCount(0);
+					assertThat(eventually(c -> !c.isPaused(), consumer)).isTrue();
+				});
+	}
+
+	@Test
+	public void exponentialBackOff() {
+		contextRunner.withPropertyValues("trigger.period=10", "trigger.initial-delay=0")
+				.run(context -> {
+					LaunchRequestConsumer consumer = consumer(context);
+					CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource(
+							context);
+					currentTaskExecutionsResource.setRunningExecutionCount(
+							currentTaskExecutionsResource.getMaximumTaskExecutions());
+					DynamicPeriodicTrigger trigger = trigger(context);
+
+					long waitTime = 0;
+					while (trigger.getDuration().compareTo(Duration.ofMillis(80)) < 0) {
+						Thread.sleep(10);
+						waitTime += 10;
+						assertThat(waitTime).isLessThan(1000);
+					}
+					assertThat(consumer.isPaused() && consumer.isRunning()).isTrue();
+				});
+	}
+
+	@Test
+	public void backoffWhenNoMessages() {
+
+		contextRunner.withPropertyValues(
+				"trigger.period=10",
+				"trigger.initial-delay=0",
+				"messageSourceDisabled=true",
+				"countDown=3")
+				.run(context -> {
+					CountDownLatch countDownLatch = countDownLatch(context);
+					CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource(
+							context);
+					DynamicPeriodicTrigger trigger = trigger(context);
+
+					assertThat(countDownLatch.await(1, TimeUnit.SECONDS)).isTrue();
+					assertThat(currentTaskExecutionsResource.getRunningExecutionCount()).isZero();
+					assertThat(trigger.getDuration()).isGreaterThanOrEqualTo(Duration.ofMillis(40));
+				});
+	}
+
+	@Test
+	public void launchRequestHasWrongPlatform() {
+		final AtomicBoolean passed = new AtomicBoolean();
+		contextRunner.withPropertyValues(
+				"trigger.period=10",
+				"trigger.initial-delay=0",
+				"autostart=false",
+				"spring.cloud.stream.bindings.input.consumer.max-attempts=1",
+				"requestWrongPlatform=true")
+				.run(context -> {
+
+					SubscribableChannel errorChannel = context.getBean("errorChannel", SubscribableChannel.class);
+					errorChannel.subscribe(message -> {
+						try {
+							assertThat(message).isInstanceOf(ErrorMessage.class);
+							ErrorMessage errorMessage = (ErrorMessage) message;
+							assertThat(errorMessage.getPayload()).isInstanceOf(Exception.class);
+							Exception exception = (Exception) message.getPayload();
+							assertThat(exception.getCause().getMessage()).isEqualTo(
+									"Task Launch request for Task foo contains deployment property 'spring.cloud.dataflow"
+											+ ".task.platformName=other' which does not match the platform configured for the Task"
+											+ " Launcher: 'default'");
+							passed.set(true);
+						}
+						catch (Exception e) {
+						}
+					});
+					LaunchRequestConsumer consumer = consumer(context);
+					consumer.start();
+					assertThat(eventually(c -> passed.get(), consumer)).isTrue();
+				});
+	}
+
+	private CurrentTaskExecutionsResource currentTaskExecutionsResource(ApplicationContext context) {
+		CurrentTaskExecutionsResource currentTaskExecutionsResource = context
+				.getBean(CurrentTaskExecutionsResource.class);
+		currentTaskExecutionsResource.setRunningExecutionCount(0);
+		currentTaskExecutionsResource.setMaximumTaskExecutions(10);
+		return currentTaskExecutionsResource;
+	}
+
+	private CountDownLatch countDownLatch(ApplicationContext context) {
+		return context.getBean(CountDownLatch.class);
+	}
+
+	private LaunchRequestConsumer consumer(ApplicationContext context) {
+		return context.getBean(LaunchRequestConsumer.class);
+	}
+
+	private DynamicPeriodicTrigger trigger(ApplicationContext context) {
+		return context.getBean(DynamicPeriodicTrigger.class);
+	}
+
+	private synchronized boolean eventually(Predicate<LaunchRequestConsumer> condition,
+			LaunchRequestConsumer consumer) {
+		final long MAX_WAIT = 1000;
+		long waitTime = 0;
+		long sleepTime = 10;
+		while (waitTime < MAX_WAIT) {
+			if (condition.test(consumer)) {
+				return true;
+			}
+			waitTime += sleepTime;
+			try {
+				Thread.sleep(sleepTime);
+			}
+			catch (InterruptedException e) {
+				Thread.interrupted();
+			}
+		}
+		return condition.test(consumer);
+	}
+
+	@SpringBootApplication
+	static class TestConfig {
+
+		private TaskOperations taskOperations;
+
+		private CurrentTaskExecutionsResource currentTaskExecutionsResource = new CurrentTaskExecutionsResource();
+
+		@Bean
+		public CurrentTaskExecutionsResource currentTaskExecutionsResource(Environment environment) {
+			currentTaskExecutionsResource.setMaximumTaskExecutions(
+					Integer.valueOf(environment.getProperty("maxExecutions", "10")));
+			currentTaskExecutionsResource.setName("default");
+			return currentTaskExecutionsResource;
+		}
+
+		@Bean
+		public CountDownLatch countDownLatch(CurrentTaskExecutionsResource resource, Environment environment) {
+			return new CountDownLatch(
+					environment.containsProperty("countDown") ? Integer.valueOf(environment.getProperty("countDown"))
+							: resource.getMaximumTaskExecutions());
+		}
+
+		@Bean
+		DataFlowOperations dataFlowOperations(CurrentTaskExecutionsResource currentTaskExecutionsResource,
+				CountDownLatch latch) {
+
+			DataFlowOperations dataFlowOperations;
+			taskOperations = mock(TaskOperations.class);
+			when(taskOperations.launch(anyString(), anyMap(), anyList(), isNull()))
+					.thenAnswer((Answer<Long>) invocation -> {
+						currentTaskExecutionsResource.setRunningExecutionCount(
+								currentTaskExecutionsResource.getRunningExecutionCount() + 1);
+						latch.countDown();
+						return Long.valueOf(currentTaskExecutionsResource.getRunningExecutionCount());
+					});
+
+			List<LauncherResource> launcherResources = new ArrayList<>();
+			LauncherResource launcherResource0 = mock(LauncherResource.class);
+			when(launcherResource0.getName()).thenReturn("default");
+			LauncherResource launcherResource1 = mock(LauncherResource.class);
+			when(launcherResource1.getName()).thenReturn("other");
+
+			when(taskOperations.currentTaskExecutions()).thenReturn(
+					Collections.singletonList(currentTaskExecutionsResource));
+			LauncherResource launcherResource = mock(LauncherResource.class);
+			when(launcherResource.getName()).thenReturn("default");
+
+			launcherResources.add(launcherResource0);
+			launcherResources.add(launcherResource1);
+
+			when(taskOperations.listPlatforms())
+					.thenReturn(PagedModel.of(launcherResources, (PagedModel.PageMetadata) null));
+
+			dataFlowOperations = mock(DataFlowOperations.class);
+			when(dataFlowOperations.taskOperations()).thenReturn(taskOperations);
+			return dataFlowOperations;
+		}
+
+		@Bean
+		public MessageSource<byte[]> testMessageSource(Environment environment, CountDownLatch countDownLatch,
+				ObjectMapper objectMapper) {
+			return () -> {
+				boolean messageSourceDisabled = Boolean.valueOf(
+						environment.getProperty("messageSourceDisabled", "false"));
+				LaunchRequest request = new LaunchRequest();
+				request.setTaskName("foo");
+				if (environment.getProperty("requestWrongPlatform", "false")
+						.equals("true")) {
+					request.getDeploymentProperties().put(LaunchRequestConsumer.TASK_PLATFORM_NAME,
+							"other");
+				}
+
+				Message<byte[]> message = null;
+
+				if (messageSourceDisabled) {
+					countDownLatch.countDown();
+				}
+				else {
+					try {
+						message = MessageBuilder.withPayload(
+								objectMapper.writeValueAsBytes(request))
+								.setHeader("contentType", "application/json")
+								.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
+										(AcknowledgmentCallback) status -> {
+										})
+								.build();
+					}
+					catch (JsonProcessingException e) {
+						throw new BeanCreationException(e.getMessage(), e);
+					}
+				}
+				return message;
+			};
+		}
+	}
+}

functions/function/tasklauncher-function/README.adoc

@@ -0,0 +1,39 @@
+# TaskLauncher Function
+
+This module provides a `Function` that uses the Data Flow REST client to launch a registered task on a configured https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-tasks[task platform].
+The client must be configured to connect to a remote Data Flow Server, including any required authentication (see Configuration Options below).
+
+## Beans for injection
+
+You can import the `TaskLauncherFunctionConfiguration` configuration in a Spring Boot application and then inject the following bean.
+
+`taskLauncherFunction`
+
+You may inject this as `TaskLauncherFunction` which implements `Function<LaunchRequest,<Optional<Long>>`.
+
+You can use `taskLauncherFunction` as a qualifier when injecting.
+
+Once injected, you can use the `apply` method of the `Function` to launch a task.
+The function takes a link:src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java[LaunchRequest] as input.
+This is a simple value object that specifies, at a minimum, the name of the task registered in Data Flow.
+Optionally, you can pass command line arguments as a `List<String>` and deployment properties as a `Map<String, String>`.
+The return value is an `Optional` containing the unique task ID of the launched instance if the launch request is successful.
+
+NOTE: This version of the tasklauncher requires Spring Cloud Data Flow version 2.4.x or higher
+
+## Configuration Options
+
+Specific properties, including the task platform name are prefixed with `tasklauncher`.
+
+For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java[TaskLauncherFunctionProperties.java]
+
+Data Flow client configuration properties are prefixed with `spring.cloud.dataflow.client`.
+Please see https://github.com/spring-cloud/spring-cloud-dataflow/blob/master/spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/config/DataFlowClientProperties.java[DataFlowClientProperties.java] for more details.
+
+## Examples
+
+See this link:src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java[test suite] for examples of how this function is used.
+
+## Other usage
+
+See this link:../../../applications/sink/tasklauncher-sink/README.adoc[README] where this function is used to create a Spring Cloud Stream application to submit task launch requests.

functions/function/tasklauncher-function/pom.xml

@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>tasklauncher-function</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>tasklauncher-function</name>
+    <description>Spring Native Function for applying filter SpEL expressions</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <properties>
+        <spring.cloud.dataflow.version>2.5.1.RELEASE</spring.cloud.dataflow.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-dataflow-rest-client</artifactId>
+            <version>${spring.cloud.dataflow.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.cloud</groupId>
+                    <artifactId>spring-cloud-scheduler-spi</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.springframework.cloud</groupId>
+                    <artifactId>spring-cloud-skipper</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+    </dependencies>
+
+</project>

functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.tasklauncher;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author David Turanski
+ **/
+
+public class LaunchRequest {
+	@JsonProperty("args")
+	private List<String> commandlineArguments = new ArrayList<>();
+	@JsonProperty("deploymentProps")
+	private Map<String, String> deploymentProperties = new HashMap<>();
+	@JsonProperty("name")
+	private String taskName;
+
+	public List<String> getCommandlineArguments() {
+		return commandlineArguments;
+	}
+
+	public void setCommandlineArguments(List<String> commandlineArguments) {
+		Assert.notNull(commandlineArguments, "'commandLineArguments' cannot be null.");
+		this.commandlineArguments = commandlineArguments;
+	}
+
+	public Map<String, String> getDeploymentProperties() {
+		return deploymentProperties;
+	}
+
+	public void setDeploymentProperties(Map<String, String> deploymentProperties) {
+		Assert.notNull(commandlineArguments, "'deploymentProperties' cannot be null.");
+		this.deploymentProperties = deploymentProperties;
+	}
+
+	public String getTaskName() {
+		return taskName;
+	}
+
+	public void setTaskName(String taskName) {
+		Assert.hasText(taskName, "'taskName' cannot be blank.");
+		this.taskName = taskName;
+	}
+}

functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.tasklauncher;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.cloud.dataflow.rest.client.TaskOperations;
+import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource;
+import org.springframework.cloud.dataflow.rest.resource.LauncherResource;
+import org.springframework.hateoas.PagedModel;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ *
+ * A {@link Function} that submits a task {@link LaunchRequest} to a Data Flow server.
+ * This will check if the Data Flow task platform is at capacity. If not, will submit the
+ * task launch request, otherwise it will return and log a warning message.
+ *
+ * @author David Turanski
+ **/
+public class TaskLauncherFunction implements Function<LaunchRequest, Optional<Long>>, InitializingBean {
+	private static final Log log = LogFactory.getLog(TaskLauncherFunction.class);
+
+	static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName";
+
+	private final TaskOperations taskOperations;
+
+	private String platformName = "default";
+
+	public TaskLauncherFunction(TaskOperations taskOperations) {
+		Assert.notNull(taskOperations, "`taskOperations` cannot be null.");
+		this.taskOperations = taskOperations;
+	}
+
+	/**
+	 *
+	 * @param launchRequest the task launch request for the Data Flow server.
+	 * @return an {@code Optional<Long>} containing the task Id if the request is accepted or
+	 * empty otherwise.
+	 */
+	@Override
+	public Optional<Long> apply(LaunchRequest launchRequest) {
+		if (platformIsAcceptingNewTasks()) {
+			return Optional.of(launchTask(launchRequest));
+		}
+		log.warn(String.format("Platform is at capacity. Did not submit task launch request for task %s.",
+				launchRequest.getTaskName()));
+		return Optional.empty();
+	}
+
+	public boolean platformIsAcceptingNewTasks() {
+
+		boolean availableForNewTasks;
+		int maximumTaskExecutions = 0;
+		int runningExecutionCount = 0;
+
+		List<String> currentPlatforms = new ArrayList<>();
+
+		boolean validPlatform = false;
+		for (CurrentTaskExecutionsResource currentTaskExecutionsResource : taskOperations.currentTaskExecutions()) {
+			if (currentTaskExecutionsResource.getName().equals(platformName)) {
+				maximumTaskExecutions = currentTaskExecutionsResource.getMaximumTaskExecutions();
+				runningExecutionCount = currentTaskExecutionsResource.getRunningExecutionCount();
+				validPlatform = true;
+			}
+			currentPlatforms.add(currentTaskExecutionsResource.getName());
+		}
+
+		// Verify for each request as configuration may have changed on server.
+		assertValidPlatform(validPlatform, currentPlatforms);
+
+		availableForNewTasks = runningExecutionCount < maximumTaskExecutions;
+		if (!availableForNewTasks) {
+			log.warn(String.format(
+					"The data Flow task platform %s has reached its concurrent task execution limit: (%d)",
+					platformName,
+					maximumTaskExecutions));
+		}
+
+		return availableForNewTasks;
+
+	}
+
+	private long launchTask(LaunchRequest request) {
+		String requestPlatformName = request.getDeploymentProperties().get(TASK_PLATFORM_NAME);
+		if (StringUtils.hasText(requestPlatformName) && !platformName.equals(requestPlatformName)) {
+			throw new IllegalStateException(
+					String.format(
+							"Task Launch request for Task %s contains deployment property '%s=%s' which does not " +
+									"match the platform configured for the Task Launcher: '%s'",
+							request.getTaskName(),
+							TASK_PLATFORM_NAME,
+							request.getDeploymentProperties().get(TASK_PLATFORM_NAME),
+							platformName));
+		}
+		log.info(String.format("Launching Task %s on platform %s", request.getTaskName(), platformName));
+		long taskId = taskOperations.launch(request.getTaskName(),
+				enrichDeploymentProperties(request.getDeploymentProperties()),
+				request.getCommandlineArguments(), null);
+		log.info(String.format("Launched Task %s - task ID is %d", request.getTaskName(), taskId));
+		return taskId;
+	}
+
+	private Map<String, String> enrichDeploymentProperties(Map<String, String> deploymentProperties) {
+		if (!deploymentProperties.containsKey(TASK_PLATFORM_NAME)) {
+			Map<String, String> enrichedProperties = new HashMap<>();
+			enrichedProperties.putAll(deploymentProperties);
+			enrichedProperties.put(TASK_PLATFORM_NAME, platformName);
+			return enrichedProperties;
+		}
+		return deploymentProperties;
+	}
+
+	public void setPlatformName(String platformName) {
+		this.platformName = platformName;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		PagedModel<LauncherResource> launchers = taskOperations.listPlatforms();
+
+		boolean validPlatform = false;
+		List<String> currentPlatforms = new ArrayList<>();
+
+		for (LauncherResource launcherResource : launchers) {
+			currentPlatforms.add(launcherResource.getName());
+			if (launcherResource.getName().equals(platformName)) {
+				validPlatform = true;
+			}
+		}
+
+		assertValidPlatform(validPlatform, currentPlatforms);
+	}
+
+	private void assertValidPlatform(boolean validPlatform, List<String> currentPlatforms) {
+		Assert.notEmpty(currentPlatforms, "The Data Flow Server has no task platforms configured");
+
+		Assert.isTrue(validPlatform, String.format(
+				"The task launcher's platform name '%s' does not match one of the Data Flow server's configured task "
+						+ "platforms: [%s].",
+				platformName, StringUtils.collectionToCommaDelimitedString(currentPlatforms)));
+	}
+}

functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.tasklauncher;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.dataflow.rest.client.DataFlowOperations;
+import org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Configuration class for the TaskLauncher Data Flow Sink.
+ *
+ * @author David Turanski
+ * @author Gunnar Hillert
+ */
+@Configuration
+@EnableConfigurationProperties({TaskLauncherFunctionProperties.class, DataFlowClientProperties.class})
+public class TaskLauncherFunctionConfiguration {
+
+	@Bean
+	public TaskLauncherFunction taskLauncherFunction(
+		DataFlowOperations dataFlowOperations, TaskLauncherFunctionProperties functionProperties) {
+
+		if (dataFlowOperations.taskOperations() == null) {
+			throw new IllegalArgumentException("The SCDF server does not support task operations");
+		}
+		TaskLauncherFunction function = new TaskLauncherFunction(dataFlowOperations.taskOperations());
+		function.setPlatformName(functionProperties.getPlatformName());
+		return function;
+	}
+}

functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.tasklauncher;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author David Turanski
+ **/
+@ConfigurationProperties("tasklauncher")
+public class TaskLauncherFunctionProperties {
+	/**
+	 * The Spring Cloud Data Flow platform to use for launching tasks.
+	 */
+	private String platformName = "default";
+
+	public String getPlatformName() {
+		return platformName;
+	}
+
+	public void setPlatformName(String platformName) {
+		this.platformName = platformName;
+	}
+}

functions/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.tasklauncher;
+
+import java.util.Collections;
+import java.util.Optional;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.dataflow.rest.client.DataFlowOperations;
+import org.springframework.cloud.dataflow.rest.client.TaskOperations;
+import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource;
+import org.springframework.cloud.dataflow.rest.resource.LauncherResource;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Profile;
+import org.springframework.hateoas.PagedModel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@SpringBootTest
+public class TaskLauncherFunctionApplicationTests {
+
+	@Autowired
+	private TaskLauncherFunction taskLauncherFunction;
+
+	@Autowired
+	private TaskOperations taskOperations;
+
+	@Test
+	public void successfulLaunch() {
+		LaunchRequest launchRequest = new LaunchRequest();
+		launchRequest.setTaskName("someTask");
+		setCurrentExecutionState(0);
+		Optional<Long> taskId = taskLauncherFunction.apply(launchRequest);
+		assertThat(taskId.isPresent()).isTrue();
+		assertThat(taskId.get()).isEqualTo(1L);
+
+		verify(taskOperations).launch("someTask",
+				Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "default"),
+				Collections.emptyList(),
+				null);
+	}
+
+	@Test
+	public void taskPlatformAtCapacity() {
+		LaunchRequest launchRequest = new LaunchRequest();
+		launchRequest.setTaskName("someTask");
+		setCurrentExecutionState(3);
+		Optional<Long> taskId = taskLauncherFunction.apply(launchRequest);
+		assertThat(taskId.isPresent()).isFalse();
+	}
+
+	@Test
+	public void platformMismatch() {
+		LaunchRequest launchRequest = new LaunchRequest();
+		launchRequest.setTaskName("someTask");
+		launchRequest
+				.setDeploymentProperties(Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "other"));
+		setCurrentExecutionState(0);
+		assertThatIllegalStateException().isThrownBy(() -> taskLauncherFunction.apply(launchRequest))
+				.withStackTraceContaining("does not match the platform configured for the Task Launcher");
+	}
+
+	private void setCurrentExecutionState(int runningExecutions) {
+		CurrentTaskExecutionsResource currentTaskExecutionsResource = new CurrentTaskExecutionsResource();
+		currentTaskExecutionsResource.setMaximumTaskExecutions(3);
+		currentTaskExecutionsResource.setRunningExecutionCount(runningExecutions);
+		currentTaskExecutionsResource.setName("default");
+		when(taskOperations.currentTaskExecutions())
+				.thenReturn(Collections.singletonList(currentTaskExecutionsResource));
+		when(taskOperations.launch(anyString(), anyMap(), anyList(), isNull())).thenReturn(1L);
+	}
+
+	@Test
+	public void noLaunchersConfigured() {
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner().withUserConfiguration(TestConfig.class);
+		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> contextRunner
+				.withPropertyValues("spring.profiles.active=nolaunchers")
+				.run(context -> context.start()))
+				.withCauseInstanceOf(BeanCreationException.class)
+				.withRootCauseInstanceOf(IllegalArgumentException.class)
+				.withStackTraceContaining("The Data Flow Server has no task platforms configured");
+	}
+
+	@Configuration
+	@Import(TaskLauncherFunctionConfiguration.class)
+	static class TestConfig {
+
+		@Bean
+		@Profile("default")
+		TaskOperations taskOperations() {
+			TaskOperations taskOperations = mock(TaskOperations.class);
+			LauncherResource launcherResource = mock(LauncherResource.class);
+			when(launcherResource.getName()).thenReturn("default");
+
+			when(taskOperations.listPlatforms()).thenReturn(PagedModel.of(
+					Collections.singletonList(launcherResource), (PagedModel.PageMetadata) null));
+			return taskOperations;
+		}
+
+		@Bean
+		@Profile("nolaunchers")
+		TaskOperations taskOperationsNoLaunchers() {
+			TaskOperations taskOperations = mock(TaskOperations.class);
+			when(taskOperations.listPlatforms()).thenReturn(PagedModel.of(
+					Collections.emptyList(), (PagedModel.PageMetadata) null));
+			return taskOperations;
+		}
+
+		@Bean
+		DataFlowOperations dataFlowOperations(TaskOperations taskOperations) {
+			DataFlowOperations dataFlowOperations = mock(DataFlowOperations.class);
+			when(dataFlowOperations.taskOperations()).thenReturn(taskOperations);
+			return dataFlowOperations;
+		}
+	}
+
+	@SpringBootApplication
+	static class TestApplication {
+	}
+}

functions/pom.xml

@@ -72,6 +72,7 @@
 		<module>function/spel-function</module>
 		<module>function/payload-converter-function</module>
 		<module>function/splitter-function</module>
+		<module>function/tasklauncher-function</module>
 
 		<module>supplier/file-supplier</module>
 		<module>supplier/ftp-supplier</module>