This section covers how you can customize the deployment of your applications. You can use a number of properties to influence settings for the applications that are deployed. Properties can be applied on a per-application basis or in the appropriate server configuration for all deployed applications.

Properties to be applied for all deployed Tasks are defined in the src/kubernetes/server/server-config-(binder).yaml file and for Streams in src/kubernetes/skipper/skipper-config-(binder).yaml. Replace (binder) with the messaging middleware you are using — for example, rabbit or kafka.

There is a limited support for using Wagon transport with maven. Currently this exists to support preemptive authentication with http based repositories and needs to be enabled manually.

Wagon based http transport is enabled by setting maven.use-wagon property to true and then preemptive authentication can be enabled per remote repository. Configuration loosely follows similar patterns found in HttpClient HTTP Wagon. At a time of writing this, documentation in Maven’s own site is slightly misleading and missing most of the possible config options.

Namespace maven.remote-repositories.<repo>.wagon.http contains all Wagon http related settings and keys directly under it maps to supported http methods, namely all, put, get and head just like in Maven’s own configuration. Under these method configurations you can then set various options like use-preemptive. Most simplest preemptive configuration sending auth header with all requests to a specified remote repository would look like:

Instead of configuring all methods it’s possible to tune settings for get and head requests only:

There are settings for use-default-headers, connection-timeout, read-timeout, request headers and HttpClient params. More about params check Wagon ConfigurationUtils.

By default, the dashboard, management, and health endpoints use HTTP as a transport. You can switch to HTTPS by adding a certificate to your configuration in application.yml, as shown in the following example:

In order to support authentication and authorization, Spring Cloud Data Flow is using OAuth 2.0 and OpenID Connect. It lets you integrate Spring Cloud Data Flow into Single Sign On (SSO) environments.

The following OAuth2 Grant Types are used:

The REST endpoints can be accessed in two ways:

You can turn on OAuth2 authentication by adding the following to application.yml or by setting environment variables. The following example shows the minimal setup needed for CloudFoundry User Account and Authentication (UAA) Server:

You can verify that basic authentication is working properly by using curl, as follows:

As a result, you should see a list of available REST endpoints.

Besides Basic Authentication, you can also provide an Access Token in order to access the REST Api. In order to make that happen, you would retrieve an OAuth2 Access Token from your OAuth2 provider first and then pass that Access Token to the REST Api using the Authorization Http header:

The preceding content deals with mostly with authentication - that is, how to assess the identity of the user. In this section we want to discuss the available authorization options - that is, who can do what.

The authorization rules are defined in dataflow-server-defaults.yml (part of the Spring Cloud Data Flow Core module).

Because the determination of security roles is environment-specific, Spring Cloud Data Flow, assigns all roles to authenticated OAuth2 users by default. The DefaultDataflowAuthoritiesExtractor class is used for that purpose.

Alternatively, Spring Cloud Data Flow can map OAuth2 scopes to Data Flow roles by setting the boolean property map-oauth-scopes for your provider to true (False is the default). For example, if your provider’s id is uaa, the property would be spring.cloud.dataflow.security.authorization.provider-role-mappings.uaa.map-oauth-scopes.

For more details, please see the chapter on Role Mappings.

Lastly, you can customize the role mapping behavior by providing your own Spring bean definition that extends Spring Cloud Data Flow’s AuthorityMapper interface. In that case, the custom bean definition takes precedence over the default one provided by Spring Cloud Data Flow.

The default scheme uses seven roles to protect the REST endpoints that Spring Cloud Data Flow exposes:

As mentioned earlier, all authorization-related default settings are specified in dataflow-server-defaults.yml, which is part of the Spring Cloud Data Flow Core Module. Nonetheless, you can override those settings, if desired - for example, in application.yml. The configuration takes the form of a YAML list (as some rules may have precedence over others). Consequently, you need to copy and paste the whole list and tailor it to your needs (as there is no way to merge lists).

The default rules are as follows:

The format of each line is the following:

where

Be mindful that the above is indeed a YAML list, not a map (thus the use of '-' dashes at the start of each line) that lives under the spring.cloud.dataflow.security.authorization.rules key.

For local deployment scenarios, we recommend using the CloudFoundry User Account and Authentication (UAA) Server, which is OpenID certified. While the UAA is used by Cloud Foundry, it is also a fully featured stand alone OAuth2 server with enterprise features such as LDAP integration.

Spring Cloud Data Flow Server offers specific set of features that can be enabled/disabled when launching. These features include all the lifecycle operations and REST endpoints (server and client implementations, including the shell and the UI) for:

One can enable and disable these features by setting the following boolean properties when launching the Data Flow server:

By default, stream (requires Skipper), and tasks are enabled and Task Scheduler is disabled by default.

The REST /about endpoint provides information on the features that have been enabled and disabled.

A relational database is used to store stream and task definitions as well as the state of executed tasks. Spring Cloud Data Flow provides schemas for H2, MySQL, Oracle, PostgreSQL, Db2, and SQL Server. The schema is automatically created when the server starts.

By default, Spring Cloud Data Flow offers an embedded instance of the H2 database. The H2 database is good for development purposes but is not recommended for production use.

The JDBC drivers for MySQL (through the MariaDB driver), PostgreSQL, SQL Server, and embedded H2 are available without additional configuration. If you are using any other database, then you need to put the corresponding JDBC driver jar on the classpath of the server.

The database properties can be passed as environment variables or command-line arguments to the Data Flow Server.

You can use the following configuration properties of the Local deployer to customize how Streams and Tasks are deployed. When deploying using the Data Flow shell, you can use the syntax deployer.<appName>.local.<deployerPropertyName>. See below for an example shell usage. These properties are also used when configuring Local Task Platforms in the Data Flow server and local platforms in Skipper for deploying Streams.

As an example, to set Java options for the time application in the ticktock stream, use the following stream deployment properties.

As a convenience, you can set the deployer.memory property to set the Java option -Xmx, as shown in the following example:

At deployment time, if you specify an -Xmx option in the deployer.<app>.local.javaOpts property in addition to a value of the deployer.<app>.local.memory option, the value in the javaOpts property has precedence. Also, the javaOpts property set when deploying the application has precedence over the Data Flow Server’s spring.cloud.deployer.local.javaOpts property.

Spring Cloud Data Flow local server is automatically configured to use RollingFileAppender for logging. The logging configuration is located on the classpath contained in a file named logback-spring.xml.

By default, the log file is configured to use:

with the logback configuration for the RollingPolicy:

To check the java.io.tmpdir for the current Spring Cloud Data Flow Server local server,

If you want to change or override any of the properties LOG_FILE, LOG_PATH, LOG_TEMP, LOG_FILE_MAX_SIZE, LOG_FILE_MAX_HISTORY and LOG_FILE_TOTAL_SIZE_CAP, please set them as system properties.

Data Flow Server delegates to the Skipper server the management of the Stream’s lifecycle. Set the configuration property spring.cloud.skipper.client.serverUri to the location of Skipper, e.g.

The configuration of show streams are deployed and to which platforms, is done by configuration of platform accounts on the Skipper server. See the documentation on platforms for more information.

The Data Flow server is responsible for deploying Tasks. Tasks that are launched by Data Flow write their state to the same database that is used by the Data Flow server. For Tasks which are Spring Batch Jobs, the job and step execution data is also stored in this database. As with streams launched by Skipper, Tasks can be launched to multiple platforms. If no platform is defined, a platform named default is created using the default values of the class LocalDeployerProperties, which is summarized in the table Local Deployer Properties

To configure new platform accounts for the local platform, provide an entry under the spring.cloud.dataflow.task.platform.local section in your application.yaml file for via another Spring Boot supported mechanism. In the following example, two local platform accounts named localDev and localDevDebug are created. The keys such as shutdownTimeout and javaOpts are local deployer properties.

When launching a task, pass the value of the platform account name using the task launch option --platformName If you do not pass a value for platformName, the value default will be used.

You can configure the Data Flow server that is running locally to deploy tasks to Cloud Foundry or Kubernetes. See the sections on Cloud Foundry Task Platform Configuration and Kubernetes Task Platform Configuration for more information.

The Spring Cloud Data Flow About Restful API result contains a display name, version, and, if specified, a URL for each of the major dependencies that comprise Spring Cloud Data Flow. The result (if enabled) also contains the sha1 and or sha256 checksum values for the shell dependency. The information that is returned for each of the dependencies is configurable by setting the following properties:

Data Flow server offers a specific set of features that you can enable or disable when launching. These features include all the lifecycle operations and REST endpoints (server, client implementations including Shell and the UI) for:

You can enable or disable these features by setting the following boolean properties when you launch the Data Flow server:

By default, all features are enabled.

The REST endpoint (/features) provides information on the enabled and disabled features.

You can use the following configuration properties of the Data Flow server’s Cloud Foundry deployer to customize how applications are deployed. When deploying with the Data Flow shell, you can use the syntax deployer.<appName>.cloudfoundry.<deployerPropertyName>. See below for an example shell usage. These properties are also used when configuring the Cloud Foundry Task platforms in the Data Flow server and and Kubernetes platforms in Skipper for deploying Streams.

Here are some examples using the Cloud Foundry deployment properties:

You can also set environment variables that specify the HTTP-based health check endpoint and timeout: SPRING_CLOUD_DATAFLOW_TASK_PLATFORM_CLOUDFOUNDRY_ACCOUNTS[default]_DEPLOYMENT_HEALTH_CHECK_ENDPOINT and SPRING_CLOUD_DATAFLOW_TASK_PLATFORM_CLOUDFOUNDRY_ACCOUNTS[default]_DEPLOYMENT_HEALTH_CHECK_TIMEOUT, respectively. These default to /health (the Spring Boot default location) and 120 seconds.

The Data Flow server is responsible for deploying Tasks. Tasks that are launched by Data Flow write their state to the same database that is used by the Data Flow server. For Tasks which are Spring Batch Jobs, the job and step execution data is also stored in this database. As with Skipper, Tasks can be launched to multiple platforms. When Data Flow is running on Cloud Foundry, a Task platfom must be defined. To configure new platform accounts that target Cloud Foundry, provide an entry under the spring.cloud.dataflow.task.platform.cloudfoundry section in your application.yaml file for via another Spring Boot supported mechanism. In the following example, two Cloud Foundry platform accounts named dev and qa are created. The keys such as memory and disk are Cloud Foundry Deployer Properties.

When launching a task, pass the value of the platform account name using the task launch option --platformName If you do not pass a value for platformName, the value default will be used.

You can configure the Data Flow server that is on Cloud Foundry to deploy tasks to Cloud Foundry or Kubernetes. See the section on Kubernetes Task Platform Configuration for more information.

To help avoid clashes with routes across spaces in Cloud Foundry, a naming strategy that provides a random prefix to a deployed application is available and is enabled by default. You can override the default configurations and set the respective properties by using cf set-env commands.

For instance, if you want to disable the randomization, you can override it by using the following command:

As an alternative to a random name or to get even more control over the hostname used by the deployed apps, you can use custom deployment properties, as the following example shows:

The preceding example binds the http app to the myhost.mydomain.com/my-path URL. Note that this example shows all of the available customization options. In practice, you can use only one or two out of the three.

Starting with version 1.2, it is possible to register and deploy Docker based apps as part of streams and tasks by using Data Flow for Cloud Foundry.

If you use Spring Boot and RabbitMQ-based Docker images, you can provide a common deployment property to facilitate binding the apps to the RabbitMQ service. Assuming your RabbitMQ service is named rabbit, you can provide the following:

For Spring Cloud Task apps, you can use something similar to the following, if you use a database service instance named mysql:

For non-Java or non-Boot applications, your Docker app must parse the VCAP_SERVICES variable in order to bind to any available services.

When deploying streams in Cloud Foundry, you can take advantage of application-specific service bindings, so not all services are globally configured for all the apps orchestrated by Spring Cloud Data Flow.

For instance, if you want to provide a mysql service binding only for the jdbc application in the following stream definition, you can pass the service binding as a deployment property:

where mysqlService is the name of the service specifically bound only to the jdbc application and the http application does not get the binding by this method.

If you have more than one service to bind, they can be passed as comma-separated items (for example: deployer.jdbc.cloudfoundry.services=mysqlService,someService).

The CloudFoundry API supports providing configuration parameters when binding a service instance. Some service brokers require or recommend binding configuration. For example, binding the Google Cloud Platform service using the CF CLI looks something like:

Likewise the NFS Volume Service supports binding configuration such as:

Starting with version 2.0, Data Flow for Cloud Foundry allows you to provide binding configuration parameters may be provided in the app level or server level cloudfoundry.services deployment property. For example, to bind to the nfs service, as above :

The format is intended to be compatible with the Data Flow DSL parser. Generally, the cloudfoundry.services deployment property accepts a comma delimited value. Since a comma is also used to separate configuration parameters, and to avoid white space issues, any item including configuration parameters must be enclosed in singe quotes. Valid values incude things like:

In addition to marketplace services, Cloud Foundry supports User-provided Services (UPS). Throughout this reference manual, regular services have been mentioned, but there is nothing precluding the use of User-provided Services as well, whether for use as the messaging middleware (for example, if you want to use an external Apache Kafka installation) or for use by some of the stream applications (for example, an Oracle Database).

Now we review an example of extracting and supplying the connection credentials from a UPS.

The following example shows a sample UPS setup for Apache Kafka:

The UPS credentials are wrapped within VCAP_SERVICES, and they can be supplied directly in the stream definition, as the following example shows.

As of Data Flow 2.0, the Spring Cloud Connector library is no longer used to create the DataSource. The library java-cfenv is now used which allows you to set Spring Boot properties to configure the connection pool.

By default, every application in Cloud Foundry starts with 1G disk quota and this can be adjusted to a default maximum of 2G. The default maximum can also be overridden up to 10G by using Pivotal Cloud Foundry’s (PCF) Ops Manager GUI.

This configuration is relevant for Spring Cloud Data Flow because every task deployment is composed of applications (typically Spring Boot uber-jar’s), and those applications are resolved from a remote maven repository. After resolution, the application artifacts are downloaded to the local Maven Repository for caching and reuse. With this happening in the background, the default disk quota (1G) can fill up rapidly, especially when we experiment with streams that are made up of unique applications. In order to overcome this disk limitation and depending on your scaling requirements, you may want to change the default maximum from 2G to 10G. Let’s review the steps to change the default maximum disk quota allocation.

Once the disk quota change has been successfully applied and assuming you have a running application, you can scale the application with a new disk_limit through the CF CLI, as the following example shows:

You can then list the applications and see the new maximum disk space, as the following example shows:

Even when configuring the Data Flow server to use 10G of space, there is the possibility of exhausting the available space on the local disk. To prevent this, jar artifacts downloaded from external sources, i.e., apps registered as http or maven resources, are automatically deleted whenever the application is deployed, whether or not the deployment request succeeds. This behavior is optimal for production environments in which container runtime stability is more critical than I/O latency incurred during deployment. In development environments deployment happens more frequently. Additionally, the jar artifact (or a lighter metadata jar) contains metadata describing application configuration properties which is used by various operations related to application configuration, more frequently performed during pre-production activities. To provide a more responsive interactive developer experience at the expense of more disk usage in pre-production environments, you can set the CloudFoundry deployer property autoDeleteMavenArtifacts to false.

If you deploy the Data Flow server by using the default port health check type, you must explicitly monitor the disk space on the server in order to avoid running out space. If you deploy the server by using the http health check type (see the next example), the Data Flow server is restarted if there is low disk space. This is due to Spring Boot’s Disk Space Health Indicator. You can configure the settings of the Disk Space Health Indicator by using the properties that have the management.health.diskspace prefix.

For version 1.7, we are investigating the use of Volume Services for the Data Flow server to store .jar artifacts before pushing them to Cloud Foundry.

The following example shows how to deploy the http health check type to an endpoint called /management/health:

Though we recommend using a Maven Artifactory for application Register a Stream App, there might be situations where one of the following alternative approaches would make sense.

By default, the Data Flow server is unsecured and runs on an unencrypted HTTP connection. You can secure your REST endpoints (as well as the Data Flow Dashboard) by enabling HTTPS and requiring clients to authenticate. For more details about securing the REST endpoints and configuring to authenticate against an OAUTH backend (UAA and SSO running on Cloud Foundry), see the security section from the core [configuration-local-security]. You can configure the security details in dataflow-server.yml or pass them as environment variables through cf set-env commands.

You must provide several pieces of configuration. These are Spring Boot @ConfigurationProperties, so you can set them as environment variables or by any other means that Spring Boot supports. The following listing is in environment variable format, as that is an easy way to get started configuring Boot applications in Cloud Foundry. Note that in the future, you will be able to deploy tasks to multiple platforms, but for 2.0.0.M1 you can deploy only to a single platform and the name must be default.

Note that you can set spring.cloud.deployer.cloudfoundry.services, spring.cloud.deployer.cloudfoundry.buildpack, or the Spring Cloud Deployer-standard spring.cloud.deployer.memory and spring.cloud.deployer.disk as part of an individual deployment request by using the deployer.<app-name> shortcut, as the following example shows:

The commands in the preceding example deploy the time source with 2048MB of memory, while the log sink uses the default 1024MB.

When you deploy a stream, you can also pass JAVA_OPTS as a deployment property, as the following example shows:

If you want to get better insights into what is happening when your streams and tasks are being deployed, you may want to turn on the following features:

You can use Spring Cloud Config Server to centralize configuration properties for Spring Boot applications. Likewise, both Spring Cloud Data Flow and the applications orchestrated by Spring Cloud Data Flow can be integrated with a configuration server to use the same capabilities.

This section discusses how to configure Spring Cloud Data Flow to connect to the PCF-Scheduler as its agent to execute tasks.

For scheduling, you must add (or update) the following environment variables in your environment:

The following sample manifest shows both environment properties configured (assuming you have a PCF-Scheduler service available with the name myscheduler):

Where the SPRING_CLOUD_SCHEDULER_CLOUDFOUNDRY_SCHEDULER_URL has the following format: scheduler.<Domain-Name> (for example, scheduler.local.pcfdev.io). Check the actual address from your PCF environment.

Data Flow server offers specific set of features that can be enabled or disabled when launching. These features include all the lifecycle operations, REST endpoints (server and client implementations including Shell and the UI) for:

You can enable or disable these features by setting the following boolean environment variables when launching the Data Flow server:

By default, all the features are enabled.

The /features REST endpoint provides information on the features that have been enabled and disabled.

You can use the following configuration properties the Kubernetes deployer to customize how Streams and Tasks are deployed. When deploying with the Data Flow shell, you can use the syntax deployer.<appName>.kubernetes.<deployerPropertyName>. These properties are also used when configuring the Kubernetes task platforms in the Data Flow server and Kubernetes platforms in Skipper for deploying Streams.

The Data Flow server is responsible for deploying Tasks. Tasks that are launched by Data Flow write their state to the same database that is used by the Data Flow server. For Tasks which are Spring Batch Jobs, the job and step execution data is also stored in this database. As with Skipper, Tasks can be launched to multiple platforms. When Data Flow is running on Kubernetes, a Task platfom must be defined. To configure new platform accounts that target Kubernetes, provide an entry under the spring.cloud.dataflow.task.platform.kubernetes section in your application.yaml file for via another Spring Boot supported mechanism. In the following example, two Kubernetes platform accounts named dev and qa are created. The keys such as memory and disk are Cloud Foundry Deployer Properties.

When launching a task, pass the value of the platform account name using the task launch option --platformName If you do not pass a value for platformName, the value default will be used.

You can configure the Data Flow server that is on Kubernetes to deploy tasks to Cloud Foundry and Kubernetes. See the section on Cloud Foundry Task Platform Configuration for more information.

The Spring Cloud Data Flow server for Kubernetes uses the spring-cloud-kubernetes module process both the ConfigMap and the secrets settings. To enable the ConfigMap support, pass in an environment variable of SPRING_CLOUD_KUBERNETES_CONFIG_NAME and set it to the name of the ConfigMap. The same is true for the secrets, where the environment variable is SPRING_CLOUD_KUBERNETES_SECRETS_NAME. To use the secrets, you also need to set SPRING_CLOUD_KUBERNETES_SECRETS_ENABLE_API to true.

The following example shows a snippet from a deployment script that sets these environment variables:

Spring Cloud Data Flow provides schemas for H2, HSQLDB, MySQL, Oracle, PostgreSQL, DB2, and SQL Server. The appropriate schema is automatically created when the server starts, provided the right database driver and appropriate credentials are in the classpath.

The JDBC drivers for MySQL (via MariaDB driver), HSQLDB, PostgreSQL, and embedded H2 are available out of the box. If you use any other database, you need to put the corresponding JDBC driver jar on the classpath of the server.

For instance, if you use MySQL in addition to a password in the secrets file, you could provide the following properties in the ConfigMap:

For PostgreSQL, you could use the following configuration:

For HSQLDB, you could use the following configuration:

You can find migration scripts for specific database types in the spring-cloud-task repo.

We recommend using the kubectl command for troubleshooting streams and tasks.

You can list all artifacts and resources used by using the following command:

You can list all resources used by a specific application or service by using a label to select resources. The following command lists all resources used by the mysql service:

You can get the logs for a specific pod by issuing the following command:

If the pod is continuously getting restarted, you can add -p as an option to see the previous log, as follows:

You can also tail or follow a log by adding an -f option, as follows:

A useful command to help in troubleshooting issues, such as a container that has a fatal error when starting up, is to use the describe command, as the following example shows:

This section covers customization of how scheduled tasks are configured. Scheduling of tasks is enabled by default in the Spring Cloud Data Flow Kubernetes Server. Properties are used to influence settings for scheduled tasks and can be configured on a global or per-schedule basis.

See KubernetesSchedulerProperties for more on the supported options.

Debugging the Spring Cloud Data Flow Kubernetes Server and included components (such as the Spring Cloud Kubernetes Deployer) is supported through the Java Debug Wire Protocol (JDWP). This section outlines an approach to manually enable debugging and another approach that uses configuration files provided with Spring Cloud Data Flow Server Kubernetes to “patch” a running deployment.

There is a Spring Shell-based client that talks to the Data Flow Server and is responsible for parsing the DSL. In turn, applications may have applications properties that rely on embedded languages, such as the Spring Expression Language.

The shell, Data Flow DSL parser, and SpEL have rules about how they handle quotes and how syntax escaping works. When combined together, confusion may arise. This section explains the rules that apply and provides examples of the most complicated situations you may encounter when all three components are involved.

A stream is defined by using a unix-inspired Pipeline syntax. The syntax uses vertical bars, also known as “pipes” to connect multiple commands. The command ls -l | grep key | less in Unix takes the output of the ls -l process and pipes it to the input of the grep key process. The output of grep in turn is sent to the input of the less process. Each | symbol connects the standard output of the command on the left to the standard input of the command on the right. Data flows through the pipeline from left to right.

In Data Flow, the Unix command is replaced by a Spring Cloud Stream application and each pipe symbol represents connecting the input and output of applications over messaging middleware, such as RabbitMQ or Apache Kafka.

Each Spring Cloud Stream application is registered under a simple name. The registration process specifies where the application can be obtained (for example, in a Maven Repository or a Docker registry). You can find out more information on how to register Spring Cloud Stream applications in this section. In Data Flow, we classify the Spring Cloud Stream applications as Sources, Processors, or Sinks.

As a simple example, consider the collection of data from an HTTP Source writing to a File Sink. Using the DSL, the stream description is:

http | file

A stream that involves some processing would be expressed as:

http | filter | transform | file

Stream definitions can be created by using the shell’s stream create command, as shown in the following example:

dataflow:> stream create --name httpIngest --definition "http | file"

The Stream DSL is passed in to the --definition command option.

The deployment of stream definitions is done through the shell’s stream deploy command.

dataflow:> stream deploy --name ticktock

The Getting Started section shows you how to start the server and how to start and use the Spring Cloud Data Flow shell.

Note that the shell calls the Data Flow Servers' REST API. For more information on making HTTP requests directly to the server, consult the REST API Guide.

The Stream Application DSL can be used to define custom binding properties for each of the Spring Cloud Stream applications. Please see the Stream Application DSL section of the microsite for more information.

or as is common when creating a Kafka Streams application,

In these cases with multiple input and output bindings, Data Flow cannot make any assumptions about the flow of data from one application to another. Therefore the developer needs to set the binding properties to 'wire up' the application. The Stream Application DSL uses a double pipe, instead of the pipe symbol, to indicate that Data Flow should not configure the binding properties of the application. Think of || as meaning 'in parallel'. For example:

dataflow:> stream create --definition "orderGeneratorApp || baristaApp || hotDrinkDeliveryApp || coldDrinkDeliveryApp" --name myCafeStream

There are four applications in this stream. The baristaApp has two output destinations, hotDrinks and coldDrinks intended to be consumed by the hotDrinkDeliveryApp and coldDrinkDeliveryApp respectively. When deploying this stream, you need to set the binding properties so that the baristaApp sends hot drink messages to the hotDrinkDeliveryApp destination and cold drink messages to the coldDrinkDeliveryApp destination. For example

If you want to use consumer groups, you will need to set the Spring Cloud Stream application property spring.cloud.stream.bindings.<channelName>.producer.requiredGroups and spring.cloud.stream.bindings.<channelName>.group on the producer and consumer applications respectively.

Another common use case for the Stream Application DSL is to deploy a http gateway application that sends a synchronous request/reply message to a Kafka or RabbitMQ application. In this case both the http gateway application and the Kafka or RabbitMQ application can be a Spring Integration application that does not make use of the Spring Cloud Stream library.

It is also possible to deploy just a single application using the Stream application DSL.

Each application takes properties to customize its behavior. As an example, the http source module exposes a port setting that allows the data ingestion port to be changed from the default value.

dataflow:> stream create --definition "http --port=8090 | log" --name myhttpstream

This port property is actually the same as the standard Spring Boot server.port property. Data Flow adds the ability to use the shorthand form port instead of server.port. One may also specify the longhand version as well, as shown in the following example:

dataflow:> stream create --definition "http --server.port=8000 | log" --name myhttpstream

This shorthand behavior is discussed more in the section on Whitelisting application properties. If you have registered application property metadata you can use tab completion in the shell after typing -- to get a list of candidate property names.

The shell provides tab completion for application properties. The shell command app info --name <appName> --type <appType> provides additional documentation for all the supported properties.

Register a versioned stream application using the app register command. You must provide a unique name, application type, and a URI that can be resolved to the app artifact. For the type, specify "source", "processor", or "sink". The version is resolved from the URI. Here are a few examples:

The application URI should conform to one the following schema formats:

If you would like to register the snapshot versions of the http and log applications built with the RabbitMQ binder, you could do the following:

If you would like to register multiple apps at one time, you can store them in a properties file where the keys are formatted as <type>.<name> and the values are the URIs.

For example, if you would like to register the snapshot versions of the http and log applications built with the RabbitMQ binder, you could have the following in a properties file (for example, stream-apps.properties):

Then to import the apps in bulk, use the app import command and provide the location of the properties file with the --uri switch, as follows:

Registering an application using --type app is the same as registering a source, processor or sink. Applications of the type app are only allowed to be used in the Stream Application DSL, which uses a comma instead of the pipe symbol in the DSL, and instructs Data Flow not to configure the Spring Cloud Stream binding properties of the application. The application that is registered using --type app does not have to be a Spring Cloud Stream app, it can be any Spring Boot application. See the Stream Application DSL introduction for more information on using this application type.

Multiple versions can be registered for the same applications (e.g. same name and type) but only one can be set as default. The default version is used for deploying Streams.

The first time an application is registered it will be marked as default. The default application version can be altered with the app default command:

The app list --id <type:name> command lists all versions for a given stream application.

The app unregister command has an optional --version parameter to specify the app version to unregister.

If a --version is not specified, the default version is unregistered.

The stream deploy necessitates default app versions to be set. The stream update and stream rollback commands though can use all (default and non-default) registered app versions.

This will create stream using the default mysource version (0.0.3). Then we can update the version to 0.0.2 like this:

An attempt to update the mysource to version 0.0.1 (not registered) will fail!

The Spring Cloud Data Flow Server exposes a full RESTful API for managing the lifecycle of stream definitions, but the easiest way to use is it is through the Spring Cloud Data Flow shell. Start the shell as described in the Getting Started section.

New streams are created with the help of stream definitions. The definitions are built from a simple DSL. For example, consider what happens if we execute the following shell command:

This defines a stream named ticktock that is based off the DSL expression time | log. The DSL uses the "pipe" symbol (|), to connect a source to a sink.

The stream info command shows useful information about the stream, as shown (with its output) in the following example:

This section describes how to deploy a Stream when the Spring Cloud Data Flow server is responsible for deploying the stream. It covers the deployment and upgrade of Streams leveraging the Skipper service. The description of how deployment properties applies to both approaches of Stream deployment.

Give the ticktock stream definition:

dataflow:> stream create --definition "time | log" --name ticktock

To deploy the stream, use the following shell command:

dataflow:> stream deploy --name ticktock

The Data Flow Server delegates to Skipper the resolution and deployment of the time and log applications.

The stream info command shows useful information about the stream, including the deployment properties:

There is an important optional command argument (called --platformName) to the stream deploy command. Skipper can be configured to deploy to multiple platforms. Skipper is pre-configured with a platform named default, which deploys applications to the local machine where Skipper is running. The default value of the command line argument --platformName is default. If you commonly deploy to one platform, when installing Skipper, you can override the configuration of the default platform. Otherwise, specify the platformName to one of the values returned by the stream platform-list command.

In the preceding example, the time source sends the current time as a message each second, and the log sink outputs it by using the logging framework. You can tail the stdout log (which has an <instance> suffix). The log files are located within the directory displayed in the Data Flow Server’s log output, as shown in the following listing:

You can also create and deploy the stream in one step by passing the --deploy flag when creating the stream, as follows:

However, it is not very common in real-world use cases to create and deploy the stream in one step. The reason is that when you use the stream deploy command, you can pass in properties that define how to map the applications onto the platform (for example, what is the memory size of the container to use, the number of each application to run, and whether to enable data partitioning features). Properties can also override application properties that were set when creating the stream. The next sections cover this feature in detail.

You can delete a stream by issuing the stream destroy command from the shell, as follows:

dataflow:> stream destroy --name ticktock

If the stream was deployed, it is undeployed before the stream definition is deleted.

Often you want to stop a stream but retain the name and definition for future use. In that case, you can undeploy the stream by name.

You can issue the deploy command at a later time to restart it.

dataflow:> stream deploy --name ticktock

Sometimes the one or more of the apps contained within a stream definition contain an invalid URI in its registration. This can caused by an invalid URI entered at app registration time or the app was removed from the repository from which it was to be drawn. To verify that all the apps contained in a stream are resolve-able, a user can use the validate command. For example:

In the example above the user validated their ticktock stream. As we see that both the source:time and sink:log are valid. Now let’s see what happens if we have a stream definition with a registered app with an invalid URI.

In this case Spring Cloud Data Flow states that the stream is invalid because source:bad-time has an invalid URI.

To update the stream, use the command stream update which takes as a command argument either --properties or --propertiesFile. There is an important new top level prefix available when using Skipper, which is version. If the Stream http | log was deployed, and the version of log which registered at the time of deployment was 1.1.0.RELEASE:

Then the following command will update the Stream to use the 1.2.0.RELEASE of the log application. Before updating the stream with the specific version of the app, we need to make sure that the app is registered with that version.

To verify the deployment properties and the updated version, we can use stream info, as shown (with its output) in the following example:

When upgrading a stream, the --force option can be used to deploy new instances of currently deployed applications even if no application or deployment properties have changed. This behavior is needed in the case when configuration information is obtained by the application itself at startup time, for example from Spring Cloud Config Server. You can specify which applications to force upgrade by using the option --app-names. If you do not specify any application names, all the applications will be force upgraded. You can specify --force and --app-names options together with --properties or --propertiesFile options.

Skipper keeps a history of the streams that were deployed. After updating a Stream, there will be a second version of the stream. You can query for the history of the versions using the command stream history --name <name-of-stream>.

Skipper keeps a “manifest” of the all the applications, their application properties, and their deployment properties after all values have been substituted. This represents the final state of what was deployed to the platform. You can view the manifest for any of the versions of a Stream by using the following command:

stream manifest --name <name-of-stream> --releaseVersion <optional-version>

If the --releaseVersion is not specified, the manifest for the last version is returned.

The following example shows the use of the manifest:

Using the command results in the following output:

The majority of the deployment and application properties were set by Data Flow to enable the applications to talk to each other and to send application metrics with identifying labels.

You can rollback to a previous version of the stream using the command stream rollback.

The optional --releaseVersion command argument adds the version of the stream. If not specified, the rollback goes to the previous stream version.

The application count is a dynamic property of the system used to specify the number of instances of applications. Please see the Application Count section of the microsite for more information.

Skipper has a simple 'red/black' upgrade strategy. It deploys the new version of the applications, using as many instances as the currently running version, and checks the /health endpoint of the application. If the health of the new application is good, then the previous application is undeployed. If the health of the new application is bad, then all new applications are undeployed and the upgrade is considered to be not successful.

The upgrade strategy is not a rolling upgrade, so if five applications of the application are running, then in a sunny-day scenario, five of the new applications are also running before the older version is undeployed.

Taps can be created at various producer endpoints in a stream. Please see the Tapping a Stream section of the microsite for more information.

When a stream is made up of multiple apps with the same name, they must be qualified with labels. Please see the Labeling Applications section of the microsite for more information.

Instead of referencing a source or sink application, you can use a named destination. Please see the Named Destinations section of the microsite for more information.

By using named destinations, you can support fan-in and fan-out use cases. Please see the Fan-in and Fan-out section of the microsite for more information.

As an example of a simple processing step, we can transform the payload of the HTTP posted data to upper case by using the following stream definition: http | transform --expression=payload.toUpperCase() | log

To create this stream enter the following command in the shell

The following example uses a shell command to post some data:

The preceding example results in an upper-case 'HELLO' in the log, as follows:

2016-06-01 09:54:37.749 INFO 80083 --- [ kafka-binder-] log.sink : HELLO

To demonstrate the data partitioning functionality, the following listing deploys a stream with Kafka as the binder:

When you review the words.log-v1-0 logs, you should see the following:

When you review the words.log-v1-1 logs, you should see the following:

This example has shown that payload splits that contain the same word are routed to the same application instance.

This example shows something a bit more complicated: swapping out the time source for something else. Another supported source type is http, which accepts data for ingestion over HTTP POSTs. Note that the http source accepts data on a different port from the Data Flow Server (default 8080). By default, the port is randomly assigned.

To create a stream using an http source but still using the same log sink, we would change the original command in the Simple Stream Processing example to the following:

Note that we do not see any other output this time until we actually post some data (by using a shell command). In order to see the randomly assigned port on which the http source is listening, run the following command:

You should see that the corresponding http source has a url property containing the host and port information on which it is listening. You are now ready to post to that url, as shown in the following example:

The stream then funnels the data from the http source to the output log implemented by the log sink, yielding output similar to the following:

We could also change the sink implementation. You could pipe the output to a file (file), to hadoop (hdfs), or to any of the other sink applications that are available. You can also define your own applications.

While Spring Cloud Task does provide a number of out-of-the-box applications (at spring-cloud-task-app-starters), most task applications require custom development. To create a custom task application:

You can register a Task App with the App Registry by using the Spring Cloud Data Flow Shell app register command. You must provide a unique name and a URI that can be resolved to the app artifact. For the type, specify "task". The following listing shows three examples:

When providing a URI with the maven scheme, the format should conform to the following:

maven://<groupId>:<artifactId>[:<extension>[:<classifier>]]:<version>

If you would like to register multiple apps at one time, you can store them in a properties file where the keys are formatted as <type>.<name> and the values are the URIs. For example, the followinng listing would be a valid properties file:

Then you can use the app import command and provide the location of the properties file by using the --uri option, as follows:

For example, if you would like to register all out-of-the-box task applications in bulk, you can do so with the following command:

You can also pass the --local option (which is TRUE by default) to indicate whether the properties file location should be resolved within the shell process itself. If the location should be resolved from the Data Flow Server process, specify --local false.

When using either app register or app import, if a task app is already registered with the provided name and version, it is not overridden by default. If you would like to override the pre-existing task app with a different uri or uri-metadata location, then include the --force option.

You can create a task Definition from a task app by providing a definition name as well as properties that apply to the task execution. Creating a task definition can be done through the RESTful API or the shell. To create a task definition by using the shell, use the task create command to create the task definition, as shown in the following example:

A listing of the current task definitions can be obtained through the RESTful API or the shell. To get the task definition list by using the shell, use the task list command.

An adhoc task can be launched through the RESTful API or the shell. To launch an ad-hoc task through the shell, use the task launch command, as shown in the following example:

When a task is launched, any properties that need to be passed as command line arguments to the task application can be set when launching the task, as follows:

Additional properties meant for a TaskLauncher itself can be passed in by using a --properties option. The format of this option is a comma-separated string of properties prefixed with app.<task definition name>.<property>. Properties are passed to TaskLauncher as application properties. It is up to an implementation to choose how those are passed into an actual task application. If the property is prefixed with deployer instead of app, it is passed to TaskLauncher as a deployment property and its meaning may be TaskLauncher implementation specific.

dataflow:>task launch mytask --properties "deployer.timestamp.custom1=value1,app.timestamp.custom2=value2"

Spring Cloud Data Flow allows a user to limit the maximum number of concurrently running tasks for each configured platform to prevent the saturation of IaaS/hardware resources. The limit is set to 20 for all supported platforms by default. If the number of concurrently running tasks on a platform instance is greater or equal to the limit, the next task launch request will fail and an error message will be returned via the RESTful API, Shell or UI. This limit can be configured for a platform instance by setting the corresponding deployer property, spring.cloud.dataflow.task.platform.<platform-type>.accounts[<account-name>].deployment.maximumConcurrentTasks property, where <account-name> is the name of a configured platform account (default if no accounts are explicitly configured). The <platform-type> refers to one of the currently supported deployers: local, cloudfoundry, or kubernetes.

The TaskLauncher implementation for each supported platform determines the number of currently executing tasks by querying the underlying platform’s runtime state if possible. The method for identifying a task varies by platform. For example, launching a task on the local host uses the LocalTaskLauncher. The LocalTaskLauncher executes a process for each launch request and keeps track of these processes in memory. In this case, we don’t query the underlying OS, as it is impractical to identify tasks this way. For Cloud Foundry, tasks are a core concept supported by its deployment model. The state of all tasks, running, completed, or failed, is available directly via the API. This means that every running task container in the account’s org and space is included in the running execution count, whether or not it was launched using Spring Cloud Data Flow, or invoking the CloudFoundryTaskLauncher directly. For Kubernetes, launching a task via the KubernetesTaskLauncher, if successful, results in a running pod which we expect to eventually complete or fail. In this environment there is generally no easy way to identify pods that correspond to a task. For this reason, we only count pods that were launched by the KubernetesTaskLauncher. Since the task launcher provides task-name label in the pod’s metadata, we filter all running pods by the presence of this label.

Once the task is launched, the state of the task is stored in a relational DB. The state includes:

A user can check the status of their task executions through the RESTful API or the shell. To display the latest task executions through the shell, use the task execution list command.

To get a list of task executions for just one task definition, add --name and the task definition name, for example task execution list --name foo. To retrieve full details for a task execution use the task execution status command with the id of the task execution, for example task execution status --id 549.

Destroying a Task Definition removes the definition from the definition repository. This can be done through the RESTful API or the shell. To destroy a task through the shell, use the task destroy command, as shown in the following example:

To destroy all tasks through the shell, use the task all destroy command as shown in the following example:

Or use the force command:

The task execution information for previously launched tasks for the definition remains in the task repository.

Sometimes the one or more of the apps contained within a task definition contain an invalid URI in its registration. This can be caused by an invalid URI entered at app registration time or the app was removed from the repository from which it was to be drawn. To verify that all the apps contained in a task are resolve-able, a user can use the validate command. For example:

In the example above the user validated their time-stamp task. As we see task:timestamp app is valid. Now let’s see what happens if we have a stream definition with a registered app with an invalid URI.

In this case Spring Cloud Data Flow states that the task is invalid because task:badtimestamp has an invalid URI.

In some cases a task that is executing on a platform may not stop because of a problem on the platform or the application business logic itself. For such cases Spring Cloud Data Flow offers the user the ability to send a request to the platform to terminate the task execution. To do this a user can submit a task execution stop for a given set of task executions. For example:

With the above command, the trigger to stop the execution of id=5 will be submitted to the underlying deployer implementation, and as a result, the operation will stop the execution of that task. When we view the result for the task execution, we will see that the task execution completed with a 0 exit code.

If a user submits a stop for a task execution that has child task executions associated with it, like a composed task, a stop request will be sent for each of the child task executions.

Composed tasks are executed through a task application called the Composed Task Runner.