You can launch a task from a stream by using one of the available task-launcher
sinks. Currently the platforms supported
via the task-launcher
sinks are
local,
Cloud Foundry, and
Yarn.
![]() | Note |
---|---|
|
A task-launcher
sink expects a message containing a TaskLaunchRequest object in its payload. From the TaskLaunchRequest
object the task-launcher
will obtain the URI of the artifact to be launched as well as the environment properties, command line arguments, deployment properties and application name to be used by the task.
The task-launcher-local can be added to the available sinks by executing the app register command as follows (for the Rabbit Binder):
app register --name task-launcher-local --type sink --uri maven://org.springframework.cloud.stream.app:task-launcher-local-sink-rabbit:jar:1.2.0.RELEASE
In the case of a maven based task that is to be launched, the task-launcher
application is responsible for downloading the artifact. You must configure the task-launcher
with the appropriate configuration of Maven Properties such as --maven.remote-repositories.repo1.url=http://repo.spring.io/libs-milestone"
to resolve artifacts, in this case against a milestone repo. Note that this repo can be different than the one used to register the task-launcher
application itself.
One way to launch a task using the task-launcher
is to use the triggertask source. The triggertask
source
will emit a message with a TaskLaunchRequest
object containing the required launch information.
The triggertask
can be added to the available sources by executing the app register command as follows (for the Rabbit Binder):
app register --type source --name triggertask --uri maven://org.springframework.cloud.stream.app:triggertask-source-rabbit:1.2.0.RELEASE
An example of this would be to launch the timestamp task once every 60 seconds, the stream to implement this would look like:
stream create foo --definition "triggertask --triggertask.uri=maven://org.springframework.cloud.task.app:timestamp-task:jar:1.2.0.RELEASE --trigger.fixed-delay=60 --triggertask.environment-properties=spring.datasource.url=jdbc:h2:tcp://localhost:19092/mem:dataflow,spring.datasource.username=sa | task-launcher-local --maven.remote-repositories.repo1.url=http://repo.spring.io/libs-release" --deploy
If you execute runtime apps
you can find the log file for the task launcher sink. Tailing that file you can find the log file for the launched tasks. The setting of triggertask.environment-properties
is so that all the task executions can be collected in the same H2 database used in the local version of the Data Flow Server. You can then see the list of task executions using the shell command task execution list
dataflow:>task execution list ╔════════════════════╤══╤════════════════════════════╤════════════════════════════╤═════════╗ ║ Task Name │ID│ Start Time │ End Time │Exit Code║ ╠════════════════════╪══╪════════════════════════════╪════════════════════════════╪═════════╣ ║timestamp-task_26176│4 │Tue May 02 12:13:49 EDT 2017│Tue May 02 12:13:49 EDT 2017│0 ║ ║timestamp-task_32996│3 │Tue May 02 12:12:49 EDT 2017│Tue May 02 12:12:49 EDT 2017│0 ║ ║timestamp-task_58971│2 │Tue May 02 12:11:50 EDT 2017│Tue May 02 12:11:50 EDT 2017│0 ║ ║timestamp-task_13467│1 │Tue May 02 12:10:50 EDT 2017│Tue May 02 12:10:50 EDT 2017│0 ║ ╚════════════════════╧══╧════════════════════════════╧════════════════════════════╧═════════╝
Another option to start a task using the task-launcher
would be to create a stream using the
Tasklaunchrequest-transform processor to translate a message payload to a TaskLaunchRequest
.
The tasklaunchrequest-transform
can be added to the available processors by executing the app register command as follows (for the Rabbit Binder):
app register --type processor --name tasklaunchrequest-transform --uri maven://org.springframework.cloud.stream.app:tasklaunchrequest-transform-processor-rabbit:1.2.0.RELEASE
For example:
stream create task-stream --definition "http --port=9000 | tasklaunchrequest-transform --uri=maven://org.springframework.cloud.task.app:timestamp-task:jar:1.2.0.RELEASE | task-launcher-local --maven.remote-repositories.repo1.url=http://repo.spring.io/libs-release"
A composed task can be launched using one of the task-launcher
sinks as discussed
here. Since we will be using
the ComposedTaskRunner
directly we will need to setup the task definitions it
will use prior to the creation of the composed task launching stream. So let’s
say that we wanted to create the following composed task definition AAA && BBB
.
The first step would be to create the task definitions. For example:
task create AAA --definition "timestamp" task create BBB --definition "timestamp"
Now that the task definitions we need for composed task definition are ready, we
need to create a stream that will launch ComposedTaskRunner
.
So in this case we will create a stream that has a trigger that will emit a message
once every 30 seconds, a transformer that will create a TaskLaunchRequest
for
each message received, and a task-launcher-local
sink that will launch a
the ComposedTaskRunner
on our local machine. The stream should look something like this:
stream create ctr-stream --definition "time --fixed-delay=30 | tasklaunchrequest-transform --uri=maven://org.springframework.cloud.task.app:composedtaskrunner-task:<current release> --command-line-arguments='--graph=AAA&&BBB --increment-instance-enabled=true --spring.datasource.url=...' | task-launcher-local"
In the example above we see that the tasklaunchrequest-transform
is establishing
2 primary components:
ComposedTaskRunner
that will be used.ComposedTaskRunner
.For now let’s focus on the configuration that is required to launch the
ComposedTaskRunner
:
ComposedTaskRunner
.
In this case it is AAA&&BBB
ComposedTaskRunner
to be unique. ComposedTaskRunner
is built using Spring Batch, and thus each
we will want a new Job Instance for each launch of the ComposedTaskRunner
. To
do this we set the increment-instance-enabled to be true
.ComposedTaskRunner
and the state of the job execution. Also this is so that the
ComposedTaskRunner
can track the state of the tasks it launched and update its state.![]() | Note |
---|---|
Releases of |