-
-
==== Schema Registry Client
-
-
-
The client-side abstraction for interacting with schema registry servers is the SchemaRegistryClient interface, with the following structure:
-
-
-
-
public interface SchemaRegistryClient {
-
- SchemaRegistrationResponse register(String subject, String format, String schema);
-
- String fetch(SchemaReference schemaReference);
-
- String fetch(Integer id);
-
-}
-
-
-
-
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:
-
-
-
-
@EnableBinding(Sink.class)
- @SpringBootApplication
- @EnableSchemaRegistryClient
- public static class AvroSinkApplication {
- ...
- }
-
-
-
-
==== Avro Schema Registry Client Message Converters
-
-
-
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.:
-
-
-
-
spring.cloud.stream.bindings.output.contentType=application/*+avro
-
-
-
-
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.
-
-
-
=== @StreamListener and Message Conversion
-
-
-
The @StreamListener annotation provides a convenient way for converting incoming messages without the need to specify the content type of an input channel.
-During the dispatching process to methods annotated with @StreamListener, a conversion will be applied automatically if the argument requires it.
-
-
-
For example, let’s consider a message with the String content {"greeting":"Hello, world"} and a content-type header of application/json is received on the input channel.
-Let us consider the following application that receives it:
-
-
-
-
public class GreetingMessage {
-
- String greeting;
-
- public String getGreeting() {
- return greeting;
- }
-
- public void setGreeting(String greeting) {
- this.greeting = greeting;
- }
-}
-
-@EnableBinding(Sink.class)
-@EnableAutoConfiguration
-public static class GreetingSink {
-
- @StreamListener(Sink.INPUT)
- public void receive(Greeting greeting) {
- // handle Greeting
- }
- }
-
-
-
-
The argument of the method will be populated automatically with the POJO containing the unmarshalled form of the JSON String.
-
-
-
== Inter-Application Communication
-
-
-
=== Connecting Multiple Application Instances
-
-
-
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:
-
-
-
-
spring.cloud.stream.bindings.output.destination=ticktock
-
-
-
-
Log Sink (that has the channel name input) will set the following property:
-
-
-
-
spring.cloud.stream.bindings.input.destination=ticktock
-
-
-
-
=== Instance Index and Instance Count
-
-
-
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.
-
-
-
-
==== Configuring Output Bindings for Partitioning
-
-
-
An output binding is configured to send partitioned data by setting one and only one of its partitionKeyExpression or partitionKeyExtractorClass properties, as well as its partitionCount property.
-For example, the following is a valid and typical configuration:
-
-
-
-
spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload.id
-spring.cloud.stream.bindings.output.producer.partitionCount=5
-
-
-
-
Based on the above example configuration, data will be sent to the target partition using the following logic.
-
-
-
A partition key’s value is calculated for each message sent to a partitioned output channel based on the partitionKeyExpression.
-The partitionKeyExpression is a SpEL expression which is evaluated against the outbound message for extracting the partitioning key.
-
-
-
If a SpEL expression is not sufficient for your needs, you can instead calculate the partition key value by setting the property partitionKeyExtractorClass to a class which implements the org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy interface.
-While the SpEL expression should usually suffice, more complex cases may use the custom implementation strategy.
-In that case, the property 'partitionKeyExtractorClass' can be set as follows:
-
-
-
-
spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=com.example.MyKeyExtractor
-spring.cloud.stream.bindings.output.producer.partitionCount=5
-
-
-
-
Once the message key is calculated, the partition selection process will determine the target partition as a value between 0 and partitionCount - 1.
-The default calculation, applicable in most scenarios, is based on the formula key.hashCode() % partitionCount.
-This can be customized on the binding, either by setting a SpEL expression to be evaluated against the 'key' (via the partitionSelectorExpression property) or by setting a org.springframework.cloud.stream.binder.PartitionSelectorStrategy implementation (via the partitionSelectorClass property).
-
-
-
The binding level properties for 'partitionSelectorExpression' and 'partitionSelectorClass' can be specified similar to the way 'partitionKeyExpression' and 'partitionKeyExtractorClass' properties are specified in the above examples.
-Additional properties can be configured for more advanced scenarios, as described in the following section.
-
-
-
===== Spring-managed custom PartitionKeyExtractorClass implementations
-
-
-
In the example above, a custom strategy such as MyKeyExtractor is instantiated by the Spring Cloud Stream directly.
-In some cases, it is necessary for such a custom strategy implementation to be created as a Spring bean, for being able to be managed by Spring, so that it can perform dependency injection, property binding, etc.
-This can be done by configuring it as a @Bean in the application context and using the fully qualified class name as the bean’s name, as in the following example.
-
-
-
-
@Bean(name="com.example.MyKeyExtractor")
-public MyKeyExtractor extractor() {
- return new MyKeyExtractor();
-}
-
-
-
-
As a Spring bean, the custom strategy benefits from the full lifecycle of a Spring bean.
-For example, if the implementation need access to the application context directly, it can make implement 'ApplicationContextAware'.
-
-
-
===== Configuring Input Bindings for Partitioning
-
-
-
An input binding (with the channel name input) is configured to receive partitioned data by setting its partitioned property, as well as the instanceIndex and instanceCount properties on the application itself, as in the following example:
-
-
-
-
spring.cloud.stream.bindings.input.consumer.partitioned=true
-spring.cloud.stream.instanceIndex=3
-spring.cloud.stream.instanceCount=5
-
-
-
-
The instanceCount value represents the total number of application instances between which the data need to be partitioned, and the instanceIndex must be a unique value across the multiple instances, between 0 and instanceCount - 1.
-The instance index helps each application instance to identify the unique partition (or, in the case of Kafka, the partition set) from which it receives data.
-It is important to set both values correctly in order to ensure that all of the data is consumed and that the application instances receive mutually exclusive datasets.
-
-
-
While a scenario which using multiple instances for partitioned data processing may be complex to set up in a standalone case, Spring Cloud Dataflow can simplify the process significantly by populating both the input and output values correctly as well as relying on the runtime infrastructure to provide information about the instance index and instance count.
-
-
-
-
Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system.
-You can do that by using the TestSupportBinder.
-This is useful especially for unit testing your microservices.
-
-
-
The TestSupportBinder allows users to interact with the bound channels and inspect what messages are sent and received by the application
-
-
-
For outbound message channels, the TestSupportBinder registers a single subscriber and retains the messages emitted by the application in a MessageCollector.
-They can be retrieved during tests and have assertions made against them.
-
-
-
The user can also send messages to inbound message channels, so that the consumer application can consume the messages.
-The following example shows how to test both input and output channels on a processor.
-
-
-
-
@RunWith(SpringJUnit4ClassRunner.class)
-@SpringApplicationConfiguration(classes = ExampleTest.MyProcessor.class)
-@IntegrationTest({"server.port=-1"})
-@DirtiesContext
-public class ExampleTest {
-
- @Autowired
- private Processor processor;
-
- @Autowired
- private BinderFactory<MessageChannel> binderFactory;
-
- @Autowired
- private MessageCollector messageCollector;
-
- @Test
- @SuppressWarnings("unchecked")
- public void testWiring() {
- Message<String> message = new GenericMessage<>("hello");
- processor.input().send(message);
- Message<String> received = (Message<String>) messageCollector.forChannel(processor.output()).poll();
- assertThat(received.getPayload(), equalTo("hello world"));
- }
-
-
- @SpringBootApplication
- @EnableBinding(Processor.class)
- public static class MyProcessor {
-
- @Autowired
- private Processor channels;
-
- @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
- public String transform(String in) {
- return in + " world";
- }
- }
-}
-
-
-
-
In the example above, we are creating an application that has an input and an output channel, bound through the Processor interface.
-The bound interface is injected into the test so we can have access to both channels.
-We are sending a message on the input channel and we are using the MessageCollector provided by Spring Cloud Stream’s test support to capture the message has been sent to the output channel as a result.
-Once we have received the message, we can validate that the component functions correctly.
-
-
-
-
Spring Cloud Stream provides a health indicator for binders.
-It is registered under the name of binders and can be enabled or disabled by setting the management.health.binders.enabled property.
-
-
-
-
-
-
To get started with creating Spring Cloud Stream applications, visit the Spring Initializr and create a new Maven project named "GreetingSource".
-Select Spring Boot {supported-spring-boot-version} in the dropdown.
-In the Search for dependencies text box type Stream Rabbit or Stream Kafka depending on what binder you want to use.
-
-
-
Next, create a new class, GreetingSource, in the same package as the GreetingSourceApplication class.
-Give it the following code:
-
-
-
-
import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.messaging.Source;
-import org.springframework.integration.annotation.InboundChannelAdapter;
-
-@EnableBinding(Source.class)
-public class GreetingSource {
-
- @InboundChannelAdapter(Source.OUTPUT)
- public String greet() {
- return "hello world " + System.currentTimeMillis();
- }
-}
-
-
-
-
The @EnableBinding annotation is what triggers the creation of Spring Integration infrastructure components.
-Specifically, it will create a Kafka connection factory, a Kafka outbound channel adapter, and the message channel defined inside the Source interface:
-
-
-
-
public interface Source {
-
- String OUTPUT = "output";
-
- @Output(Source.OUTPUT)
- MessageChannel output();
-
-}
-
-
-
-
The auto-configuration also creates a default poller, so that the greet() method will be invoked once per second.
-The standard Spring Integration @InboundChannelAdapter annotation sends a message to the source’s output channel, using the return value as the payload of the message.
-
-
-
To test-drive this setup, run a Kafka message broker.
-An easy way to do this is to use a Docker image:
-
-
-
-
# On OS X
-$ docker run -p 2181:2181 -p 9092:9092 --env ADVERTISED_HOST=`docker-machine ip \`docker-machine active\`` --env ADVERTISED_PORT=9092 spotify/kafka
-
-# On Linux
-$ docker run -p 2181:2181 -p 9092:9092 --env ADVERTISED_HOST=localhost --env ADVERTISED_PORT=9092 spotify/kafka
-
-
-
-
Build the application:
-
-
-
-
The consumer application is coded in a similar manner.
-Go back to Initializr and create another project, named LoggingSink.
-Then create a new class, LoggingSink, in the same package as the class LoggingSinkApplication and with the following code:
-
-
-
-
import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.messaging.Sink;
-
-@EnableBinding(Sink.class)
-public class LoggingSink {
-
- @StreamListener(Sink.INPUT)
- public void log(String message) {
- System.out.println(message);
- }
-}
-
-
-
-
Build the application:
-
-
-
-
To connect the GreetingSource application to the LoggingSink application, each application must share the same destination name.
-Starting up both applications as shown below, you will see the consumer application printing "hello world" and a timestamp to the console:
-
-
-
-
cd GreetingSource
-java -jar target/GreetingSource-0.0.1-SNAPSHOT.jar --spring.cloud.stream.bindings.output.destination=mydest
-
-cd LoggingSink
-java -jar target/LoggingSink-0.0.1-SNAPSHOT.jar --server.port=8090 --spring.cloud.stream.bindings.input.destination=mydest
-
-
-
-
(The different server port prevents collisions of the HTTP port used to service the Spring Boot Actuator endpoints in the two applications.)
-
-
-
The output of the LoggingSink application will look something like the following:
-
-
-
-
[ main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 8090 (http)
-[ main] com.example.LoggingSinkApplication : Started LoggingSinkApplication in 6.828 seconds (JVM running for 7.371)
-hello world 1458595076731
-hello world 1458595077732
-hello world 1458595078733
-hello world 1458595079734
-hello world 1458595080735
-
-
-
-
= Binder Implementations
-
-
-
== Apache Kafka Binder
-
-
-
-
-
For using the Apache Kafka binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:
-
-
-
-
<dependency>
- <groupId>org.springframework.cloud</groupId>
- <artifactId>spring-cloud-stream-binder-kafka</artifactId>
-</dependency>
-
-
-
-
Alternatively, you can also use the Spring Cloud Stream Kafka Starter.
-
-
-
-
<dependency>
- <groupId>org.springframework.cloud</groupId>
- <artifactId>spring-cloud-starter-stream-kafka</artifactId>
-</dependency>
-
-
-
-
== Apache Kafka Binder Overview
-
-
-
A simplified diagram of how the Apache Kafka binder operates can be seen below.
-
-
-
-
-
-
Figure 9. Kafka Binder
-
-
-
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.
-
-
-
== Configuration Options
-
-
-
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 docs.
-
-
-
=== Kafka Binder Properties
-
-
-
-- spring.cloud.stream.kafka.binder.brokers
--
-
A list of brokers to which the Kafka binder will connect.
-
-
-- spring.cloud.stream.kafka.binder.defaultBrokerPort
--
-
brokers allows hosts specified with or without port information (e.g., host1,host2:port2).
-This sets the default port when no port is configured in the broker list.
-
-
-- spring.cloud.stream.kafka.binder.zkNodes
--
-
A list of ZooKeeper nodes to which the Kafka binder can connect.
-
-
-- spring.cloud.stream.kafka.binder.defaultZkPort
--
-
zkNodes allows hosts specified with or without port information (e.g., host1,host2:port2).
-This sets the default port when no port is configured in the node list.
-
-
-- spring.cloud.stream.kafka.binder.configuration
--
-
Key/Value map of client properties (both producers and consumer) passed to all clients created by the binder.
-Due to the fact that these properties will be used by both producers and consumers, usage should be restricted to common properties, especially security settings.
-
-
-- spring.cloud.stream.kafka.binder.headers
--
-
The list of custom headers that will be transported by the binder.
-
-
-- spring.cloud.stream.kafka.binder.offsetUpdateTimeWindow
--
-
The frequency, in milliseconds, with which offsets are saved.
-Ignored if 0.
-
-
-- spring.cloud.stream.kafka.binder.offsetUpdateCount
--
-
The frequency, in number of updates, which which consumed offsets are persisted.
-Ignored if 0.
-Mutually exclusive with offsetUpdateTimeWindow.
-
-
-- spring.cloud.stream.kafka.binder.requiredAcks
--
-
The number of required acks on the broker.
-
-
-- spring.cloud.stream.kafka.binder.minPartitionCount
--
-
Effective only if autoCreateTopics or autoAddPartitions is set.
-The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data.
-It can be superseded by the partitionCount setting of the producer or by the value of instanceCount * concurrency settings of the producer (if either is larger).
-
-
-- spring.cloud.stream.kafka.binder.replicationFactor
--
-
The replication factor of auto-created topics if autoCreateTopics is active.
-
-
-- spring.cloud.stream.kafka.binder.autoCreateTopics
--
-
If set to true, the binder will create new topics automatically.
-If set to false, the binder will rely on the topics being already configured.
-In the latter case, if the topics do not exist, the binder will fail to start.
-Of note, this setting is independent of the auto.topic.create.enable setting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings.
-
-
-- spring.cloud.stream.kafka.binder.autoAddPartitions
--
-
If set to true, the binder will create add new partitions if required.
-If set to false, the binder will rely on the partition size of the topic being already configured.
-If the partition count of the target topic is smaller than the expected value, the binder will fail to start.
-
-
-- spring.cloud.stream.kafka.binder.socketBufferSize
--
-
Size (in bytes) of the socket buffer to be used by the Kafka consumers.
-
-
-
-
-
-
=== Kafka Consumer Properties
-
-
-
The following properties are available for Kafka consumers only and
-must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.consumer..
-
-
-
-- autoRebalanceEnabled
--
-
When true, topic partitions will be automatically rebalanced between the members of a consumer group.
-When false, each consumer will be assigned a fixed set of partitions based on spring.cloud.stream.instanceCount and spring.cloud.stream.instanceIndex.
-This requires both spring.cloud.stream.instanceCount and spring.cloud.stream.instanceIndex properties to be set appropriately on each launched instance.
-The property spring.cloud.stream.instanceCount must typically be greater than 1 in this case.
-
-
-- autoCommitOffset
--
-
Whether to autocommit offsets when a message has been processed.
-If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header will be present in the inbound message.
-Applications may use this header for acknowledging messages.
-See the examples section for details.
-When this property is set to false, Kafka binder will set the ack mode to org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL.
-
-
-- autoCommitOnError
--
-
Effective only if autoCommitOffset is set to true.
-If set to false it suppresses auto-commits for messages that result in errors, and will commit only for successful messages, allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.
-If set to true, it will always auto-commit (if auto-commit is enabled).
-If not set (default), it effectively has the same value as enableDlq, auto-committing erroneous messages if they are sent to a DLQ, and not committing them otherwise.
-
-
-- recoveryInterval
--
-
The interval between connection recovery attempts, in milliseconds.
-
-
-- resetOffsets
--
-
Whether to reset offsets on the consumer to the value provided by startOffset.
-
-
-- startOffset
--
-
The starting offset for new groups, or when resetOffsets is true.
-Allowed values: earliest, latest.
-If the consumer group is set explicitly for the consumer 'binding' (via spring.cloud.stream.bindings.<channelName>.group), then 'startOffset' is set to earliest; otherwise it is set to latest for the anonymous consumer group.
-
-
Default: null (equivalent to earliest).
-
-- enableDlq
--
-
When set to true, it will send enable DLQ behavior for the consumer.
-Messages that result in errors will be forwarded to a topic named error.<destination>.<group>.
-This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
-
-
-- configuration
--
-
Map with a key/value pair containing generic Kafka consumer properties.
-
-
-
-
-
-
=== Kafka Producer Properties
-
-
-
The following properties are available for Kafka producers only and
-must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.producer..
-
-
-
-- bufferSize
--
-
Upper limit, in bytes, of how much data the Kafka producer will attempt to batch before sending.
-
-
-- sync
--
-
Whether the producer is synchronous.
-
-
-- batchTimeout
--
-
How long the producer will wait before sending in order to allow more messages to accumulate in the same batch.
-(Normally the producer does not wait at all, and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency.
-
-
-- configuration
--
-
Map with a key/value pair containing generic Kafka producer properties.
-
-
-
-
-
-
-
-
-
-
-
