Visit the Spring Initializr. This is where we’ll generate our LoggingConsumer application.

In the Dependencies start typing 'stream' and Cloud Stream option should pop up. Select it. Now start typing either 'kafka' or 'rabbit'. Basically this is where you are choosing what messaging midleware this application will be bound to. Choose the one you have already installed and/or feel more comfortable with installing/running. Also, as you can see from the Initilaizer screen there are few other options you can choose. For example, you can choose Gradle as your build tool instead of the default Maven. With the Dependencies selected the only other thing you have to identify is the application name - logging-consumer. Your configuration screeen should now contain the following:

Click on Generate Project button. This will donwload the zipped version of the generated project to your hard drive. Unzip it and you’re ready for Step Two.

Here you simply import the project into your IDE of choice. Please keep in mind that dependening on the IDE you may need to follow a specific import procedures. For example depending on how the project was generated (Maven or Gradle) you may need to follow specific import procedure (e.g., in Eclipse/STS: File → Import → Maven → Existing Maven Project).

Ones imported the project must have no errors of any kind and src/main/java should also contain com.example.loggingconsumer.LoggingConsumerApplication.

Technically at this point you can just run the application’s main class since it’s already a valid Spring Boot application, but it does not do anything, so let’s add some code.

Modify the com.example.loggingconsumer.LoggingConsumerApplication to look as follows:

As you can see from the above:

This is it, we now have a fully functional Spring Cloud Stream application that does something. From here for simplicity we’ll assume RabbitMQ was selected in step one. Assuming you have RabbitMQ installed and running, start the application by simply running its main method.

You should see following output:

Go to RabbitMQ management console or any other RabbitMQ client and simply send message to input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg (NOTE: the anonymous.CbMIwdkJSBO1ZoPDOtHtCg part represents the group name and is generated and will be different in your environment. For something more predictable you can use explicit group name via spring.cloud.stream.bindings.input.group=hello).

The contents of the message should be JSON representation of Person class, so let’s send this:

And in your console you should see:

You can also build/package your application into a boot jar (i.e., ./mvnw clean install) and run the built JAR using java -jar command.

That is all!

A Spring Cloud Stream application consists of a middleware-neutral core. The application communicates with the outside world through input and output channels injected into it by Spring Cloud Stream. Channels are connected to external brokers through middleware-specific Binder implementations.

Spring Cloud Stream provides Binder implementations for Kafka and Rabbit MQ. Spring Cloud Stream also includes a TestSupportBinder, which leaves a channel unmodified so that tests can interact with channels directly and reliably assert on what is received. You can use the extensible API to write your own Binder.

Spring Cloud Stream uses Spring Boot for configuration, and the Binder abstraction makes it possible for a Spring Cloud Stream application to be flexible in how it connects to middleware. For example, deployers can dynamically choose, at runtime, the destinations (e.g., the Kafka topics or RabbitMQ exchanges) to which channels connect. Such configuration can be provided through external configuration properties and in any form supported by Spring Boot (including application arguments, environment variables, and application.yml or application.properties files). In the sink example from the Introducing Spring Cloud Stream section, setting the application property spring.cloud.stream.bindings.input.destination to raw-sensor-data will cause it to read from the raw-sensor-data Kafka topic, or from a queue bound to the raw-sensor-data RabbitMQ exchange.

Spring Cloud Stream automatically detects and uses a binder found on the classpath. You can easily use different types of middleware with the same code: just include a different binder at build time. For more complex use cases, you can also package multiple binders with your application and have it choose the binder, and even whether to use different binders for different channels, at runtime.

Communication between applications follows a publish-subscribe model, where data is broadcast through shared topics. This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.

Data reported by sensors to an HTTP endpoint is sent to a common destination named raw-sensor-data. From the destination, it is independently processed by a microservice application that computes time-windowed averages and by another microservice application that ingests the raw data into HDFS. In order to process the data, both applications declare the topic as their input at runtime.

The publish-subscribe communication model reduces the complexity of both the producer and the consumer, and allows new applications to be added to the topology without disruption of the existing flow. For example, downstream from the average-calculating application, you can add an application that calculates the highest temperature values for display and monitoring. You can then add another application that interprets the same flow of averages for fault detection. Doing all communication through shared topics rather than point-to-point queues reduces coupling between microservices.

While the concept of publish-subscribe messaging is not new, Spring Cloud Stream takes the extra step of making it an opinionated choice for its application model. By using native middleware support, Spring Cloud Stream also simplifies use of the publish-subscribe model across different platforms.

While the publish-subscribe model makes it easy to connect applications through shared topics, the ability to scale up by creating multiple instances of a given application is equally important. When doing this, different instances of an application are placed in a competing consumer relationship, where only one of the instances is expected to handle a given message.

Spring Cloud Stream models this behavior through the concept of a consumer group. (Spring Cloud Stream consumer groups are similar to and inspired by Kafka consumer groups.) Each consumer binding can use the spring.cloud.stream.bindings.<channelName>.group property to specify a group name. For the consumers shown in the following figure, this property would be set as spring.cloud.stream.bindings.<channelName>.group=hdfsWrite or spring.cloud.stream.bindings.<channelName>.group=average.

All groups which subscribe to a given destination receive a copy of published data, but only one member of each group receives a given message from that destination. By default, when a group is not specified, Spring Cloud Stream assigns the application to an anonymous and independent single-member consumer group that is in a publish-subscribe relationship with all other consumer groups.

Two types of consumer are supported:

Prior to version 2.0, only asynchronous consumers were supported, where a message is delivered as soon as it is available (and there is a thread available to process it).

You might want to use a synchronous consumer when you wish to control the rate at which messages are processed.

Spring Cloud Stream provides support for partitioning data between multiple instances of a given application. In a partitioned scenario, the physical communication medium (e.g., the broker topic) is viewed as being structured into multiple partitions. One or more producer application instances send data to multiple consumer application instances and ensure that data identified by common characteristics are processed by the same consumer instance.

Spring Cloud Stream provides a common abstraction for implementing partitioned processing use cases in a uniform fashion. Partitioning can thus be used whether the broker itself is naturally partitioned (e.g., Kafka) or not (e.g., RabbitMQ).

Partitioning is a critical concept in stateful processing, where it is critical, for either performance or consistency reasons, to ensure that all related data is processed together. For example, in the time-windowed average calculation example, it is important that all measurements from any given sensor are processed by the same application instance.

A producer is any component that sends messages to a channel. The channel can be bound to an external message broker via a Binder implementation for that broker. When invoking the bindProducer() method, the first parameter is the name of the destination within the broker, the second parameter is the local channel instance to which the producer will send messages, and the third parameter contains properties (such as a partition key expression) to be used within the adapter that is created for that channel.

A consumer is any component that receives messages from a channel. As with a producer, the consumer’s channel can be bound to an external message broker. When invoking the bindConsumer() method, the first parameter is the destination name, and a second parameter provides the name of a logical group of consumers. Each group that is represented by consumer bindings for a given destination receives a copy of each message that a producer sends to that destination (i.e., publish-subscribe semantics). If there are multiple consumer instances bound using the same group name, then messages will be load-balanced across those consumer instances so that each message sent by a producer is consumed by only a single consumer instance within each group (i.e., queueing semantics).

The Binder SPI consists of a number of interfaces, out-of-the box utility classes and discovery strategies that provide a pluggable mechanism for connecting to external middleware.

The key point of the SPI is the Binder interface which is a strategy for connecting inputs and outputs to external middleware.

The interface is parameterized, offering a number of extension points:

A typical binder implementation consists of the following

Spring Cloud Stream relies on implementations of the Binder SPI to perform the task of connecting channels to message brokers. Each Binder implementation typically connects to one type of messaging system.

When multiple binders are present on the classpath, the application must indicate which binder is to be used for each channel binding. Each binder configuration contains a META-INF/spring.binders, which is a simple properties file:

Similar files exist for the other provided binder implementations (e.g., Kafka), and custom binder implementations are expected to provide them, as well. The key represents an identifying name for the binder implementation, whereas the value is a comma-separated list of configuration classes that each contain one and only one bean definition of type org.springframework.cloud.stream.binder.Binder.

Binder selection can either be performed globally, using the spring.cloud.stream.defaultBinder property (e.g., spring.cloud.stream.defaultBinder=rabbit) or individually, by configuring the binder on each channel binding. For instance, a processor application (that has channels with the names input and output for read/write respectively) which reads from Kafka and writes to RabbitMQ can specify the following configuration:

By default, binders share the application’s Spring Boot auto-configuration, so that one instance of each binder found on the classpath will be created. If your application should connect to more than one broker of the same type, you can specify multiple binder configurations, each with different environment settings.

For example, this is the typical configuration for a processor application which connects to two RabbitMQ broker instances:

Since version 2.0 Spring Cloud Stream supports visualization and control of the Bindings via Actuator endpoints.

Once the above prerequisites are satisfied you should see the following in the logs when application is started:

To visualize current bindings simply access the following URL:

or

…​if you want to visualize a single binding named 'myBindingName'

You can also stop, start, pause and resume individual binding by posting to the same URL while providing state argument as JSON.

For example,

The following properties are available when creating custom binder configurations. They must be prefixed with spring.cloud.stream.binders.<configurationName>.

Binding properties are supplied using the format spring.cloud.stream.bindings.<channelName>.<property>=<value>. The <channelName> represents the name of the channel being configured (e.g., output for a Source).

To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format spring.cloud.stream.default.<property>=<value>.

In what follows, we indicate where we have omitted the spring.cloud.stream.bindings.<channelName>. prefix and focus just on the property name, with the understanding that the prefix will be included at runtime.

Besides the channels defined via @EnableBinding, Spring Cloud Stream allows applications to send messages to dynamically bound destinations. This is useful, for example, when the target destination needs to be determined at runtime. Applications can do so by using the BinderAwareChannelResolver bean, registered automatically by the @EnableBinding annotation.

The property 'spring.cloud.stream.dynamicDestinations' can be used for restricting the dynamic destination names to a set known beforehand (whitelisting). If the property is not set, any destination can be bound dynamically.

The BinderAwareChannelResolver can be used directly as in the following example, in which a REST controller uses a path variable to decide the target channel.

After starting the application on the default port 8080, when sending the following data:

The destinations 'customers' and 'orders' are created in the broker (for example: exchange in case of Rabbit or topic in case of Kafka) with the names 'customers' and 'orders', and the data is published to the appropriate destinations.

The BinderAwareChannelResolver is a general purpose Spring Integration DestinationResolver and can be injected in other components. For example, in a router using a SpEL expression based on the target field of an incoming JSON message.

The Router Sink Application uses this technique to create the destinations on-demand.

If the channel names are known in advance, you can configure the producer properties as with any other destination. Alternatively, if you register a NewBindingCallback<> bean, it will be invoked just before the binding is created. The callback takes the generic type of the extended producer properties used by the binder; it has one method:

The following is an example using the RabbitMQ binder:

Data transformation is one of the core features of any message-driven microservice architecture. Given that in Spring Cloud Stream, such data is represented as a Spring Message, such message may have to be transformed to a desired shape/size before reaching its destination. This is required for two reasons:

1. To convert the contents of the incoming message to match the signature of the application-provided handler.

2. To convert the contents of the outgoing message to the wire format.

The wire format is typically byte[] (i.e., Kafka and Rabbit binders), but is governed by the binder implementation.

In Spring Cloud Stream, message transformation is accomplished with a org.springframework.messaging.converter.MessageConverter.

To better understand the mechanics and the necessity behind content-type negotiation let’s look at the very simple use case using the following message handler as an example. Also let’s assume that this is the only handler in the application (no internal pipeline) for simplicity.

The above handler expects Person type as an argument and will produce String type as an output. In order for the framework to succeed in passing the incoming Message as an argument to this handler it has to somehow transform the payload of the Message from the wire format to Person type. In other words the framework must locate and apply the appropriate MessageConverter. To accomplish that the framework needs some instructions from the user. One of these instructions is already provided by the signature of the handler method itself (Person type), so in theory, that should and in some cases is enough, but for the majority of the use cases in order to select the appropriate MessageConverter the framework needs an additional piece of information. That missing piece is contentType.

Spring Cloud Stream provides three simple mechanisms to define contentType and they all come with precedence order:

1. HEADER - the contentType can be communicated through the Message itself. By simply providing contentType header you are declaring the content type to use to locate and apply the appropriate MessageConverter.

2. BINDING - the contentType can be set per destination binding via spring.cloud.stream.bindings.input.content-type property. NOTE: the segment input in the property name corresponds to the actual name of the destination which is “input” in our case. This approach allows one to declare per-binding the content type to use to locate and apply the appropriate MessageConverter.

3. DEFAULT - in the event contentType is not present in the Message header and/or binding, the default application/json content type will be used to locate and apply the appropriate MessageConverter.

As mentioned, the above also demonstrates the order of precedence in the event there is a tie. For example, header provided content type takes precedence over any other content type. The same applies for content type set per binding which essentially allows one to override the default content type. But it also provides a sensible default which was determined from the community feedback.

Another reason for making application/json the default stems from the interoperability requirements driven by distributed microservices architectures where producer and consumer not only run in different JVMs, but can also run on different non-JVM platforms.

Once the non-void handler method returns and unless the return value is already a Message, the new Message is constructed with return vlaue as the payload while inheriting headers from the input Message less the ones defined/filtered by SpringIntegrationProperties.messageHandlerNotPropagatedHeaders. By default, there is only one header set there - contentType. This means that the new Message will not have contentType header set, thus ensuring that the contentType can evolve. You can always opt out to returning a Message from the handler method where you can inject any header you wish.

If there is an internal pipeline the Message is sent to the next handler going through the same process of conversion, or if there is no internal pipeline or you’ve reached the end of it the Message is sent back to the output destination.

As it was mentioned earlier the framework already provides a stack of MessageConverters to handle most common use cases. Below is the ordered list of provided MessageConverters.

In the event no appropriate converter is found the framework will throw an exception at which point you should check your code and configfuration and ensure you didn’t miss anything (i.e., provide contentType via binding or header). However, most likely you are dealing with some uncommon case (custom contentType perhaps) and the current stack of provided MessageConverters doesn’t know how to convert. And if that’s the case you can add custom MessageConverter.

Spring Cloud Stream exposes a mechanism to define and register additional MessageConverters. All you need to do is implement org.springframework.messaging.converter.MessageConverter, confiure it as @Bean and annotate it with @StreamMessageConverter and it will be added to the existing stack of MessageConverters. The @StreamMessageConverter qualifier annotation is to avoid picking up other converters that may be present on the Application Context.

Here is an example of creating a message converter bean to support new content type application/bar:

Spring Cloud Stream also provides support for Avro-based converters and schema evolution. See the specific section for details.

The spring-cloud-stream-schema module contains two types of message converters that can be used for Apache Avro serialization:

The AvroSchemaMessageConverter supports serializing and deserializing messages either using a predefined schema or by using the schema information available in the class (either reflectively, or contained in the SpecificRecord). If the target type of the conversion is a GenericRecord, then a schema must be set.

For using it, you can simply add it to the application context, optionally specifying one ore more MimeTypes to associate it with. The default MimeType is application/avro.

Here is an example of configuring it in a sink application registering the Apache Avro MessageConverter, without a predefined schema:

Conversely, here is an application that registers a converter with a predefined schema, to be found on the classpath:

In order to understand the schema registry client converter, we will describe the schema registry support first.

Most serialization models, especially the ones that aim for portability across different platforms and languages, rely on a schema that describes how the data is serialized in the binary payload. In order to serialize the data and then to interpret it, both the sending and receiving sides must have access to a schema that describes the binary format. In certain cases, the schema can be inferred from the payload type on serialization, or from the target type on deserialization, but in a lot of cases applications benefit from having access to an explicit schema that describes the binary data format. A schema registry allows you to store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format. A schema is referenceable as a tuple consisting of:

Spring Cloud Stream provides a schema registry server implementation. In order to use it, you can simply add the spring-cloud-stream-schema-server artifact to your project and use the @EnableSchemaRegistryServer annotation, adding the schema registry server REST controller to your application. This annotation is intended to be used with Spring Boot web applications, and the listening port of the server is controlled by the server.port setting. The spring.cloud.stream.schema.server.path setting can be used to control the root path of the schema server (especially when it is embedded in other applications). The spring.cloud.stream.schema.server.allowSchemaDeletion boolean setting enables the deletion of schema. By default this is disabled.

The schema registry server uses a relational database to store the schemas. By default, it uses an embedded database. You can customize the schema storage using the Spring Boot SQL database and JDBC configuration options.

A Spring Boot application enabling the schema registry looks as follows:

The client-side abstraction for interacting with schema registry servers is the SchemaRegistryClient interface, with the following structure:

Spring Cloud Stream provides out of the box implementations for interacting with its own schema server, as well as for interacting with the Confluent Schema Registry.

A client for the Spring Cloud Stream schema registry can be configured using the @EnableSchemaRegistryClient as follows:

For Spring Boot applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream will auto-configure an Apache Avro message converter that uses the schema registry client for schema management. This eases schema evolution, as applications that receive messages can get easy access to a writer schema that can be reconciled with their own reader schema.

For outbound messages, the MessageConverter will be activated if the content type of the channel is set to application/*+avro, e.g.:

During the outbound conversion, the message converter will try to infer the schemas of the outbound messages based on their type and register them to a subject based on the payload type using the SchemaRegistryClient. If an identical schema is already found, then a reference to it will be retrieved. If not, the schema will be registered and a new version number will be provided. The message will be sent with a contentType header using the scheme application/[prefix].[subject].v[version]+avro, where prefix is configurable and subject is deduced from the payload type.

For example, a message of the type User may be sent as a binary payload with a content type of application/vnd.user.v2+avro, where user is the subject and 2 is the version number.

When receiving messages, the converter will infer the schema reference from the header of the incoming message and will try to retrieve it. The schema will be used as the writer schema in the deserialization process.

To better understand how Spring Cloud Stream registers and resolves new schemas, as well as its use of Avro schema comparison features, we will provide two separate subsections below: one for the registration, and one for the resolution of schemas.

While Spring Cloud Stream makes it easy for individual Spring Boot applications to connect to messaging systems, the typical scenario for Spring Cloud Stream is the creation of multi-application pipelines, where microservice applications send data to each other. You can achieve this scenario by correlating the input and output destinations of adjacent applications.

Supposing that a design calls for the Time Source application to send data to the Log Sink application, you can use a common destination named ticktock for bindings within both applications.

Time Source (that has the channel name output) will set the following property:

Log Sink (that has the channel name input) will set the following property:

When scaling up Spring Cloud Stream applications, each instance can receive information about how many other instances of the same application exist and what its own instance index is. Spring Cloud Stream does this through the spring.cloud.stream.instanceCount and spring.cloud.stream.instanceIndex properties. For example, if there are three instances of a HDFS sink application, all three instances will have spring.cloud.stream.instanceCount set to 3, and the individual applications will have spring.cloud.stream.instanceIndex set to 0, 1, and 2, respectively.

When Spring Cloud Stream applications are deployed via Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. By default, spring.cloud.stream.instanceCount is 1, and spring.cloud.stream.instanceIndex is 0.

In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (e.g., the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances.

The intent behind the test binder superseding all the other binders on the classpath is to make it easy to test your applications without making changes to your production dependencies. In some cases (e.g. integration tests) it is useful to use the actual production binders instead, and that requires disabling the test binder autoconfiguration. In order to do so, you can exclude the org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration class using one of the Spring Boot autoconfiguration exclusion mechanisms, as in the following example.

When autoconfiguration is disabled, the test binder is available on the classpath, and its defaultCandidate property is set to false, so that it does not interfere with the regular user configuration. It can be referenced under the name test e.g.:

On CloudFoundry services are usually exposed via a special environment variable called VCAP_SERVICES.

When configuring your binder connections, you can use the values from an environment variable as explained on the dataflow cloudfoundry server docs.

To use Apache Kafka binder all you need is to add spring-cloud-stream-binder-kafka as a dependency to your Spring Cloud Stream application. Below is a Maven example:

Alternatively, you can also use the Spring Cloud Stream Kafka Starter.

A simplified diagram of how the Apache Kafka binder operates can be seen below.

The Apache Kafka Binder implementation maps each destination to an Apache Kafka topic. The consumer group maps directly to the same Apache Kafka concept. Partitioning also maps directly to Apache Kafka partitions as well.

The binder currently uses the Apache Kafka kafka-clients 1.0.0 jar and is designed to be used with a broker at least that version. This client can communicate with older brokers (refer to the Kafka documentation), but certain features may not be available. For example, with versions earlier than 0.11.x.x, native headers are not supported. Also, 0.11.x.x does not support the autoAddPartitions property.

This section contains the configuration options used by the Apache Kafka binder.

For common configuration options and properties pertaining to binder, refer to the core documentation.

Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too. See Message Channel Binders and Error Channels for more information.

The payload of the ErrorMessage for a send failure is a KafkaSendFailureException with properties:

There is no automatic handling of producer exceptions (such as sending to a Dead-Letter queue); you can consume these exceptions with your own Spring Integration flow.

Kafka binder module exposes the following metrics:

spring.cloud.stream.binder.kafka.someGroup.someTopic.lag - this metric indicates how many messages have not been yet consumed from given binder’s topic by given consumer group. For example if the value of the metric spring.cloud.stream.binder.kafka.myGroup.myTopic.lag is 1000, then consumer group myGroup has 1000 messages to waiting to be consumed from topic myTopic. This metric is particularly useful to provide auto-scaling feedback to PaaS platform of your choice.

Because it can’t be anticipated how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. If the reason for the dead-lettering is transient, you may wish to route the messages back to the original topic. However, if the problem is a permanent issue, that could cause an infinite loop. The following spring-boot application is an example of how to route those messages back to the original topic, but moves them to a third "parking lot" topic after three attempts. The application is simply another spring-cloud-stream application that reads from the dead-letter topic. It terminates when no messages are received for 5 seconds.

The examples assume the original destination is so8400out and the consumer group is so8400.

There are several considerations.

Apache Kafka supports topic partitioning natively.

Sometimes it is advantageous to send data to specific partitions, for example when you want to strictly order message processing - all messages for a particular customer should go to the same partition.

The following illustrates how to configure the producer and consumer side:

Since partitions are natively handled by Kafka, no special configuration is needed on the consumer side. Kafka will allocate partitions across the instances.

You can add instances as needed; Kafka will rebalance the partition allocations. If the instance count (or instance count * concurrency) exceeds the number of partitions, some consumers will be idle.

For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:

Spring Cloud Stream’s Apache Kafka support also includes a binder implementation designed explicitly for Apache Kafka Streams binding. With this native integration, a Spring Cloud Stream "processor" application can directly use the Apache Kafka Streams APIs in the core business logic.

Kafka Streams binder implementation builds on the foundation provided by the Kafka Streams in Spring Kafka project.

As part of this native integration, the high-level Streams DSL provided by the Kafka Streams API is available for use in the business logic, too.

An early version of the Processor API support is available as well.

As noted early-on, Kafka Streams support in Spring Cloud Stream strictly only available for use in the Processor model. A model in which the messages read from an inbound topic, business processing can be applied, and the transformed messages can be written to an outbound topic. It can also be used in Processor applications with a no-outbound destination.

This section contains the configuration options used by the Kafka Streams binder.

For common configuration options and properties pertaining to binder, refer to the core documentation.

For use cases that requires multiple incoming KStream objects or a combination of KStream and KTable objects, the Kafka Streams binder provides multiple bindings support.

Let’s see it in action.

Kafka Streams allow outbound data to be split into multiple topics based on some predicates. The Kafka Streams binder provides support for this feature without compromising the programming model exposed through StreamListener in the end user application.

You can write the application in the usual way as demonstrated above in the word count example. However, when using the branching feature, you are required to do a few things. First, you need to make sure that your return type is KStream[] instead of a regular KStream. Second, you need to use the SendTo annotation containing the output bindings in the order (see example below). For each of these output bindings, you need to configure destination, content-type etc., complying with the standard Spring Cloud Stream expectations.

Here is an example:

Properties:

Similar to message-channel based binder applications, the Kafka Streams binder adapts to the out-of-the-box content-type conversions without any compromise.

It is typical for Kafka Streams operations to know the type of SerDe’s used to transform the key and value correctly. Therefore, it may be more natural to rely on the SerDe facilities provided by the Apache Kafka Streams library itself at the inbound and outbound conversions rather than using the content-type conversions offered by the framework. On the other hand, you might be already familiar with the content-type conversion patterns provided by the framework, and that, you’d like to continue using for inbound and outbound conversions.

Both the options are supported in the Kafka Streams binder implementation.

Apache Kafka Streams provide the capability for natively handling exceptions from deserialization errors. For details on this support, please see this Out of the box, Apache Kafka Streams provide two kinds of deserialization exception handlers - logAndContinue and logAndFail. As the name indicates, the former will log the error and continue processing the next records and the latter will log the error and fail. LogAndFail is the default deserialization exception handler.

As part of the public Kafka Streams binder API, we expose a class called QueryableStoreRegistry. You can access this as a Spring bean in your application. An easy way to get access to this bean from your application is to "autowire" the bean in your application.

Once you gain access to this bean, then you can query for the particular state-store that you are interested. See below.

For using the RabbitMQ binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:

Alternatively, you can also use the Spring Cloud Stream RabbitMQ Starter.

A simplified diagram of how the RabbitMQ binder operates can be seen below.

The RabbitMQ Binder implementation maps each destination to a TopicExchange (by default). For each consumer group, a Queue will be bound to that TopicExchange. Each consumer instance have a corresponding RabbitMQ Consumer instance for its group’s Queue. For partitioned producers/consumers the queues are suffixed with the partition index and use the partition index as routing key. For anonymous consumers (no group property) an auto-delete queue is used, with a randomized unique name.

Using the optional autoBindDlq option, you can configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange DLX as well as routing infrastructure). By default, the dead letter queue has the name of the destination, appended with .dlq. If retry is enabled (maxAttempts > 1) failed messages will be delivered to the DLQ after retries are exhausted. If retry is disabled (maxAttempts = 1), you should set requeueRejected to false (default) so that a failed message will be routed to the DLQ, instead of being requeued. In addition, republishToDlq causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the x-exception-stacktrace header. This option does not need retry enabled; you can republish a failed message after just one attempt. Starting with version 1.2, you can configure the delivery mode of republished messages; see property republishDeliveryMode.

See RabbitMQ Binder Properties for more information about these properties.

The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue). Some options are described in Dead-Letter Queue Processing.

Starting with version 2.0, the RabbitMessageChannelBinder sets the RabbitTemplate.userPublisherConnection property to true so that the non-transactional producers will avoid dead locks on consumers which can happen if cached connections are blocked because of Memory Alarm on Broker.

This section contains settings specific to the RabbitMQ Binder and bound channels.

For general binding configuration options and properties, please refer to the Spring Cloud Stream core documentation.

Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too. See Message Channel Binders and Error Channels for more information.

With rabbitmq, there are two types of send failures:

The latter is rare; quoting the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".

As well as enabling producer error channels as described in Message Channel Binders and Error Channels, the RabbitMQ binder will only send messages to the channels if the connection factory is appropriately configured:

When using spring boot configuration for the connection factory, set properties:

The payload of the ErrorMessage for a returned message is a ReturnedAmqpMessageException with properties:

For negatively acknowledged confirms, the payload is a NackedAmqpMessageException with properties:

There is no automatic handling of these exceptions (such as sending to a Dead-Letter queue); you can consume these exceptions with your own Spring Integration flow.

Because it can’t be anticipated how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. If the reason for the dead-lettering is transient, you may wish to route the messages back to the original queue. However, if the problem is a permanent issue, that could cause an infinite loop. The following spring-boot application is an example of how to route those messages back to the original queue, but moves them to a third "parking lot" queue after three attempts. The second example utilizes the RabbitMQ Delayed Message Exchange to introduce a delay to the requeued message. In this example, the delay increases for each attempt. These examples use a @RabbitListener to receive messages from the DLQ, you could also use RabbitTemplate.receive() in a batch process.

The examples assume the original destination is so8400in and the consumer group is so8400.

RabbitMQ does not support partitioning natively.

Sometimes it is advantageous to send data to specific partitions, for example when you want to strictly order message processing - all messages for a particular customer should go to the same partition.

The RabbitMessageChannelBinder provides partitioning by binding a queue for each partition to the destination exchange.

The following illustrates how to configure the producer and consumer side:

This configuration provisions a topic exchange:

and these queues bound to that exchange:

with these bindings:

To build the source you will need to install JDK 1.7.

The build uses the Maven wrapper so you don’t have to install a specific version of Maven. To enable the tests for Redis, Rabbit, and Kafka bindings you should have those servers running before building. See below for more information on running the servers.

The main build command is

You can also add '-DskipTests' if you like, to avoid running the tests.

The projects that require middleware generally include a docker-compose.yml, so consider using Docker Compose to run the middeware servers in Docker containers. See the README in the scripts demo repository for specific instructions about the common cases of mongo, rabbit and redis.

There is a "full" profile that will generate documentation.

If you don’t have an IDE preference we would recommend that you use Spring Tools Suite or Eclipse when working with the code. We use the m2eclipe eclipse plugin for maven support. Other IDEs and tools should also work without issue.

Before we accept a non-trivial patch or pull request we will need you to sign the contributor’s agreement. Signing the contributor’s agreement does not grant anyone commit rights to the main repository, but it does mean that we can accept your contributions, and you will get an author credit if we do. Active contributors might be asked to join the core team, and given the ability to merge pull requests.

None of these is essential for a pull request, but they will all help. They can also be added after the original pull request but before a merge.