diff --git a/multi/multi__configuration_options.html b/multi/multi__configuration_options.html index bf1934e40..9215fadcd 100644 --- a/multi/multi__configuration_options.html +++ b/multi/multi__configuration_options.html @@ -1,6 +1,6 @@ - 8. Configuration Options
8. Configuration Options
Prev Part II. Reference Guide Next

8. Configuration Options

Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders. + 8. Configuration Options

8. Configuration Options
Prev Part I. Preface Next

8. Configuration Options

Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders. Some binders let additional binding properties support middleware-specific features.

Configuration options can be provided to Spring Cloud Stream applications through any mechanism supported by Spring Boot. This includes application arguments, environment variables, and YAML or .properties files.

8.1 Binding Service Properties

These properties are exposed via org.springframework.cloud.stream.config.BindingServiceProperties

spring.cloud.stream.instanceCount

The number of deployed instances of an application. Must be set for partitioning on the producer side. Must be set on the consumer side when using RabbitMQ and with Kafka if autoRebalanceEnabled=false.

Default: 1.

spring.cloud.stream.instanceIndex
The instance index of the application: A number from 0 to instanceCount - 1. @@ -124,4 +124,4 @@ It has one method:

"myDLQ");
     };
-}
[Note]Note

If you need to support dynamic destinations with multiple binder types, use Object for the generic type and cast the extended argument as needed.

Prev Up Next
7. Binders Home 9. Content Type Negotiation
\ No newline at end of file +}
[Note]Note

If you need to support dynamic destinations with multiple binder types, use Object for the generic type and cast the extended argument as needed.

Prev Up Next
7. Binders Home 9. Content Type Negotiation
\ No newline at end of file diff --git a/multi/multi__health_indicator.html b/multi/multi__health_indicator.html index 46604ca66..10e45418c 100644 --- a/multi/multi__health_indicator.html +++ b/multi/multi__health_indicator.html @@ -1,6 +1,6 @@ - 13. Health Indicator
13. Health Indicator
Prev Part II. Reference Guide Next

13. Health Indicator

Spring Cloud Stream provides a health indicator for binders. + 13. Health Indicator

13. Health Indicator
Prev Part I. Preface Next

13. Health Indicator

Spring Cloud Stream provides a health indicator for binders. It is registered under the name binders and can be enabled or disabled by setting the management.health.binders.enabled property.

To enable health check you first need to enable both "web" and "actuator" by including its dependencies (see Section 3.2.1, “Both Actuator and Web Dependencies Are Now Optional”)

If management.health.binders.enabled is not set explicitly by the application, then management.health.defaults.enabled is matched as true and the binder health indicators are enabled. If you want to disable health indicator completely, then you have to set management.health.binders.enabled to false.

You can use Spring Boot actuator health endpoint to access the health indicator - /actuator/health. By default, you will only receive the top level application status when you hit the above endpoint. @@ -9,4 +9,4 @@ you can do so by setting property management.health.binder In this case, the health indicator infrastructure from Spring Boot will still pick up these custom beans. Even if you are not disabling the binder health indicators, you can still enhance the health checks by providing your own HealthIndicator beans in addition to the out of the box health checks.

When you have multiple binders in the same application, health indicators are enabled by default unless the application turns them off by setting management.health.binders.enabled to false. In this case, if the user wants to disable health check for a subset of the binders, then that should be done by setting management.health.binders.enabled to false in the multi binder configurations’s environment. -See Connecting to Multiple Systems for details on how environment specific properties can be provided.

Prev Up Next
12. Testing Home 14. Metrics Emitter
\ No newline at end of file +See Connecting to Multiple Systems for details on how environment specific properties can be provided.

Prev Up Next
12. Testing Home 14. Metrics Emitter
\ No newline at end of file diff --git a/multi/multi__inter_application_communication.html b/multi/multi__inter_application_communication.html index f7f871b5e..75d749855 100644 --- a/multi/multi__inter_application_communication.html +++ b/multi/multi__inter_application_communication.html @@ -1,6 +1,6 @@ - 11. Inter-Application Communication
11. Inter-Application Communication
Prev Part II. Reference Guide Next

11. Inter-Application Communication

Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:

11.1 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. + 11. Inter-Application Communication

11. Inter-Application Communication
Prev Part I. Preface Next

11. Inter-Application Communication

Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:

11.1 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.

Suppose a design calls for the Time Source application to send data to the Log Sink application. You could use a common destination named ticktock for bindings within both applications.

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

spring.cloud.stream.bindings.output.destination=ticktock

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

spring.cloud.stream.bindings.input.destination=ticktock

11.2 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 have spring.cloud.stream.instanceCount set to 3, and the individual applications have spring.cloud.stream.instanceIndex set to 0, 1, and 2, respectively.

When Spring Cloud Stream applications are deployed through Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. @@ -34,4 +34,4 @@ With Kafka, if autoRebalanceEnabled is autoRebalanceEnabled is set to false, the instanceCount and instanceIndex are used by the binder to determine which partition(s) the instance subscribes to (you must have at least as many partitions as there are instances). The binder allocates the partitions instead of Kafka. This might be useful if you want messages for a particular partition to always go to the same instance. -When a binder configuration requires them, 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 in 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 and by letting you rely on the runtime infrastructure to provide information about the instance index and instance count.

Prev Up Next
10. Schema Evolution Support Home 12. Testing
\ No newline at end of file +When a binder configuration requires them, 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 in 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 and by letting you rely on the runtime infrastructure to provide information about the instance index and instance count.

Prev Up Next
10. Schema Evolution Support Home 12. Testing
\ No newline at end of file diff --git a/multi/multi__main_concepts.html b/multi/multi__main_concepts.html index 0294dbb63..90be8765d 100644 --- a/multi/multi__main_concepts.html +++ b/multi/multi__main_concepts.html @@ -1,9 +1,9 @@ - 5. Main Concepts
5. Main Concepts
Prev Part II. Reference Guide Next

5. Main Concepts

Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications. + 5. Main Concepts

5. Main Concepts
Prev Part I. Preface Next

5. Main Concepts

Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications. This section gives an overview of the following:

5.1 Application Model

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.

Figure 5.1. Spring Cloud Stream Application

SCSt with binder

5.1.1 Fat JAR

Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing. +Channels are connected to external brokers through middleware-specific Binder implementations.

Figure 5.1. Spring Cloud Stream Application

SCSt with binder

5.1.1 Fat JAR

Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing. To run a Spring Cloud Stream application in production, you can create an executable (or “fat”) JAR by using the standard Spring Boot tooling provided for Maven or Gradle. See the Spring Boot Reference Guide for more details.

5.2 The Binder Abstraction

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 also 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. @@ -13,7 +13,7 @@ In the sink example from the

5.3 Persistent Publish-Subscribe Support

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.

Figure 5.2. Spring Cloud Stream Publish-Subscribe

SCSt sensors

Data reported by sensors to an HTTP endpoint is sent to a common destination named raw-sensor-data. +This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.

Figure 5.2. Spring Cloud Stream Publish-Subscribe

SCSt sensors

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 (Hadoop Distributed File System). 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 lets new applications 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. @@ -23,7 +23,7 @@ By using native middleware support, Spring Cloud Stream also simplifies use of t When doing so, 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.

Figure 5.3. Spring Cloud Stream Consumer Groups

SCSt groups

All groups that 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. +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.

Figure 5.3. Spring Cloud Stream Consumer Groups

SCSt groups

All groups that 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.

5.5 Consumer Types

Two types of consumer are supported:

  • Message-driven (sometimes referred to as Asynchronous)
  • Polled (sometimes referred to as Synchronous)

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

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

5.5.1 Durability

Consistent with the opinionated application model of Spring Cloud Stream, consumer group subscriptions are durable. That is, a binder implementation ensures that group subscriptions are persistent and that, once at least one subscription for a group has been created, the group receives messages, even if they are sent while all applications in the group are stopped.

[Note]Note

Anonymous subscriptions are non-durable by nature. For some binder implementations (such as RabbitMQ), it is possible to have non-durable group subscriptions.

In general, it is preferable to always specify a consumer group when binding an application to a given destination. @@ -31,5 +31,5 @@ When scaling up a Spring Cloud Stream application, you must specify a consumer g Doing so prevents the application’s instances from receiving duplicate messages (unless that behavior is desired, which is unusual).

5.6 Partitioning Support

Spring Cloud Stream provides support for partitioning data between multiple instances of a given application. In a partitioned scenario, the physical communication medium (such as 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 (for example, Kafka) or not (for example, RabbitMQ).

Figure 5.4. Spring Cloud Stream Partitioning

SCSt partitioning

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.

[Note]Note

To set up a partitioned processing scenario, you must configure both the data-producing and the data-consuming ends.

Prev Up Next
4. Introducing Spring Cloud Stream Home 6. Programming Model
\ No newline at end of file +Partitioning can thus be used whether the broker itself is naturally partitioned (for example, Kafka) or not (for example, RabbitMQ).

Figure 5.4. Spring Cloud Stream Partitioning

SCSt partitioning

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.

[Note]Note

To set up a partitioned processing scenario, you must configure both the data-producing and the data-consuming ends.

Prev Up Next
4. Introducing Spring Cloud Stream Home 6. Programming Model
\ No newline at end of file diff --git a/multi/multi__programming_model.html b/multi/multi__programming_model.html index 7324b6c25..ee5a1664b 100644 --- a/multi/multi__programming_model.html +++ b/multi/multi__programming_model.html @@ -1,6 +1,6 @@ - 6. Programming Model
6. Programming Model
Prev Part II. Reference Guide Next

6. Programming Model

To understand the programming model, you should be familiar with the following core concepts:

  • Destination Binders: Components responsible to provide integration with the external messaging systems.
  • Destination Bindings: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).
  • Message: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).
SCSt overview

6.1 Destination Binders

Destination Binders are extension components of Spring Cloud Stream responsible for providing the necessary configuration and implementation to facilitate + 6. Programming Model

6. Programming Model
Prev Part I. Preface Next

6. Programming Model

To understand the programming model, you should be familiar with the following core concepts:

  • Destination Binders: Components responsible to provide integration with the external messaging systems.
  • Destination Bindings: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).
  • Message: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).
SCSt overview

6.1 Destination Binders

Destination Binders are extension components of Spring Cloud Stream responsible for providing the necessary configuration and implementation to facilitate integration with external messaging systems. This integration is responsible for connectivity, delegation, and routing of messages to and from producers and consumers, data type conversion, invocation of the user code, and more.

Binders handle a lot of the boiler plate responsibilities that would otherwise fall on your shoulders. However, to accomplish that, the binder still needs @@ -250,7 +250,7 @@ If the service activator throws a RequeueCurrentMessageExc The error handling comes in two flavors:

  • application: The error handling is done within the application (custom error handler).
  • system: The error handling is delegated to the binder (re-queue, DL, and others). Note that the techniques are dependent on binder implementation and the capability of the underlying messaging middleware.

Spring Cloud Stream uses the Spring Retry library to facilitate successful message processing. See Section 6.4.3, “Retry Template” for more details. However, when all fails, the exceptions thrown by the message handlers are propagated back to the binder. At that point, binder invokes custom error handler or communicates -the error back to the messaging system (re-queue, DLQ, and others).

6.4.1 Application Error Handling

There are two types of application-level error handling. Errors can be handled at each binding subscription or a global handler can handle all the binding subscription errors. Let’s review the details.

Figure 6.1. A Spring Cloud Stream Sink Application with Custom and Global Error Handlers

custom vs global error channels

For each input binding, Spring Cloud Stream creates a dedicated error channel with the following semantics <destinationName>.errors.

[Note]Note

The <destinationName> consists of the name of the binding (such as input) and the name of the group (such as myGroup).

Consider the following:

spring.cloud.stream.bindings.input.group=myGroup
@StreamListener(Sink.INPUT) // destination name 'input.myGroup'
+the error back to the messaging system (re-queue, DLQ, and others).
6.4.1 Application Error Handling
There are two types of application-level error handling. Errors can be handled at each binding subscription or a global handler can handle all the binding subscription errors. Let’s review the details.
Figure 6.1. A Spring Cloud Stream Sink Application with Custom and Global Error Handlers
custom vs global error channels

For each input binding, Spring Cloud Stream creates a dedicated error channel with the following semantics <destinationName>.errors.
[Note]Note
The <destinationName> consists of the name of the binding (such as input) and the name of the group (such as myGroup).
Consider the following:
spring.cloud.stream.bindings.input.group=myGroup
@StreamListener(Sink.INPUT) // destination name 'input.myGroup'
 public void handle(Person value) {
 	throw new RuntimeException("BOOM!");
 }
@@ -392,4 +392,4 @@ The Publisher in the following example still uses R
         e -> e.poller(p -> p.fixedDelay(1)))
         .toReactivePublisher();
   }
-}
Prev Up Next
5. Main Concepts Home 7. Binders
\ No newline at end of file +}
Prev Up Next
5. Main Concepts Home 7. Binders
\ No newline at end of file diff --git a/multi/multi__samples.html b/multi/multi__samples.html index ef81664f0..5fbd677de 100644 --- a/multi/multi__samples.html +++ b/multi/multi__samples.html @@ -1,3 +1,3 @@ - 15. Samples
15. Samples
Prev Part II. Reference Guide Next

15. Samples

For Spring Cloud Stream samples, see the spring-cloud-stream-samples repository on GitHub.

15.1 Deploying Stream Applications on CloudFoundry

On CloudFoundry, services are usually exposed through 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 Cloud Foundry Server docs.

Prev Up Next
14. Metrics Emitter Home Part III. Appendices
\ No newline at end of file + 15. Samples
15. Samples
Prev Part I. Preface 

15. Samples

For Spring Cloud Stream samples, see the spring-cloud-stream-samples repository on GitHub.

15.1 Deploying Stream Applications on CloudFoundry

On CloudFoundry, services are usually exposed through 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 Cloud Foundry Server docs.

Prev Up 
14. Metrics Emitter Home 
\ No newline at end of file diff --git a/multi/multi__testing.html b/multi/multi__testing.html index 94e376802..7a456a38e 100644 --- a/multi/multi__testing.html +++ b/multi/multi__testing.html @@ -1,6 +1,6 @@ - 12. Testing
12. Testing
Prev Part II. Reference Guide Next

12. Testing

Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system. + 12. Testing

12. Testing
Prev Part I. Preface Next

12. Testing

Spring Cloud Stream provides support for testing your microservice applications without connecting to a messaging system. You can do that by using the TestSupportBinder provided by the spring-cloud-stream-test-support library, which can be added as a test dependency to the application, as shown in the following example:

   <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-test-support</artifactId>
@@ -53,4 +53,4 @@ To do so, you can exclude the org.springframework.cloud.st
         public String transform(String in) {
             return in + " world";
         }
-    }

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, as shown in the following example:

spring.cloud.stream.defaultBinder=test

Prev Up Next
11. Inter-Application Communication Home 13. Health Indicator
\ No newline at end of file + }

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, as shown in the following example:

spring.cloud.stream.defaultBinder=test

Prev Up Next
11. Inter-Application Communication Home 13. Health Indicator
\ No newline at end of file diff --git a/multi/multi__whats_new_in_2_0.html b/multi/multi__whats_new_in_2_0.html index a9581e4a8..36eff59a2 100644 --- a/multi/multi__whats_new_in_2_0.html +++ b/multi/multi__whats_new_in_2_0.html @@ -1,6 +1,6 @@ - 3. What’s New in 2.0?
3. What’s New in 2.0?
Prev Part I. Preface Next

3. What’s New in 2.0?

Spring Cloud Stream introduces a number of new features, enhancements, and changes. The following sections outline the most notable ones:

3.1 New Features and Components

  • Polling Consumers: Introduction of polled consumers, which lets the application control message processing rates. + 3. What’s New in 2.0?
    3. What’s New in 2.0?
    Prev Part I. Preface Next

    3. What’s New in 2.0?

    Spring Cloud Stream introduces a number of new features, enhancements, and changes. The following sections outline the most notable ones:

    3.1 New Features and Components

    • Polling Consumers: Introduction of polled consumers, which lets the application control message processing rates. See “Section 6.3.5, “Using Polled Consumers”” for more details. You can also read this blog post for more details.
    • Micrometer Support: Metrics has been switched to use Micrometer. MeterRegistry is also provided as a bean so that custom applications can autowire it to capture custom metrics. @@ -22,11 +22,11 @@ See the “this blog post contains more detail.

      • All message conversion is now handled only by MessageConverter objects.
      • We introduced the @StreamMessageConverter annotation to provide custom MessageConverter objects.
      • We introduced the default Content Type as application/json, which needs to be taken into consideration when migrating 1.3 application or operating in the mixed mode (that is, 1.3 producer → 2.0 consumer).
      • Messages with textual payloads and a contentType of text/…​ or …​/json are no longer converted to Message<String> for cases where the argument type of the provided MessageHandler can not be determined (that is, public void handle(Message<?> message) or public void handle(Object payload)). Furthermore, a strong argument type may not be enough to properly convert messages, so the contentType header may be used as a supplement by some MessageConverters.

    3.3 Notable Deprecations

    As of version 2.0, the following items have been deprecated:

    3.3.1 Java Serialization (Java Native and Kryo)

    JavaSerializationMessageConverter and KryoMessageConverter remain for now. However, we plan to move them out of the core packages and support in the future. The main reason for this deprecation is to flag the issue that type-based, language-specific serialization could cause in distributed environments, where Producers and Consumers may depend on different JVM versions or have different versions of supporting libraries (that is, Kryo). -We also wanted to draw the attention to the fact that Consumers and Producers may not even be Java-based, so polyglot style serialization (i.e., JSON) is better suited.

    3.3.2 Deprecated Classes and Methods

    The following is a quick summary of notable deprecations. See the corresponding javadoc for more details.

    • SharedChannelRegistry. Use SharedBindingTargetRegistry.
    • Bindings. +We also wanted to draw the attention to the fact that Consumers and Producers may not even be Java-based, so polyglot style serialization (i.e., JSON) is better suited.

    3.3.2 Deprecated Classes and Methods

    The following is a quick summary of notable deprecations. See the corresponding {spring-cloud-stream-javadoc-current}[javadoc] for more details.

    • SharedChannelRegistry. Use SharedBindingTargetRegistry.
    • Bindings. Beans qualified by it are already uniquely identified by their type — for example, provided Source, Processor, or custom bindings:
    public interface Sample {
 	String OUTPUT = "sampleOutput";
 
 	@Output(Sample.OUTPUT)
 	MessageChannel output();
 }
    • HeaderMode.raw. Use none, headers or embeddedHeaders
    • ProducerProperties.partitionKeyExtractorClass in favor of partitionKeyExtractorName and ProducerProperties.partitionSelectorClass in favor of partitionSelectorName. -This change ensures that both components are Spring configured and managed and are referenced in a Spring-friendly way.
    • BinderAwareRouterBeanPostProcessor. While the component remains, it is no longer a BeanPostProcessor and will be renamed in the future.
    • BinderProperties.setEnvironment(Properties environment). Use BinderProperties.setEnvironment(Map<String, Object> environment).
    Prev Up Next
    2. Quick Start Home Part II. Reference Guide
    \ No newline at end of file +This change ensures that both components are Spring configured and managed and are referenced in a Spring-friendly way.
  • BinderAwareRouterBeanPostProcessor. While the component remains, it is no longer a BeanPostProcessor and will be renamed in the future.
  • BinderProperties.setEnvironment(Properties environment). Use BinderProperties.setEnvironment(Map<String, Object> environment).
Prev Up Next
2. Quick Start Home 4. Introducing Spring Cloud Stream
\ No newline at end of file diff --git a/multi/multi_content-type-management.html b/multi/multi_content-type-management.html index dd75b1c6c..ebf9f0318 100644 --- a/multi/multi_content-type-management.html +++ b/multi/multi_content-type-management.html @@ -1,6 +1,6 @@ - 9. Content Type Negotiation
9. Content Type Negotiation
Prev Part II. Reference Guide Next

9. Content Type Negotiation

Data transformation is one of the core features of any message-driven microservice architecture. Given that, in Spring Cloud Stream, such data + 9. Content Type Negotiation

9. Content Type Negotiation
Prev Part I. Preface Next

9. Content Type Negotiation

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, a message may have to be transformed to a desired shape or 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[] (that is true for the Kafka and Rabbit binders), but it is governed by the binder implementation.

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

[Note]Note

As a supplement to the details to follow, you may also want to read the following blog post.

9.1 Mechanics

To better understand the mechanics and the necessity behind content-type negotiation, we take a look at a very simple use case by using the following message handler as an example:

@StreamListener(Processor.INPUT)
 @SendTo(Processor.OUTPUT)
 public String handle(Person person) {..}
[Note]Note

For simplicity, we assume that this is the only handler in the application (we assume there is no internal pipeline).

The handler shown in the preceding example expects a Person object as an argument and produces a String type as an output. @@ -75,4 +75,4 @@ Consequently, custom MessageConverter implementatio return (payload instanceof Bar ? payload : new Bar((byte[]) payload)); } }

Spring Cloud Stream also provides support for Avro-based converters and schema evolution. -See “Chapter 10, Schema Evolution Support” for details.

Prev Up Next
8. Configuration Options Home 10. Schema Evolution Support
\ No newline at end of file +See “Chapter 10, Schema Evolution Support” for details.

Prev Up Next
8. Configuration Options Home 10. Schema Evolution Support
\ No newline at end of file diff --git a/multi/multi_schema-evolution.html b/multi/multi_schema-evolution.html index 7a5f95add..3e28eff72 100644 --- a/multi/multi_schema-evolution.html +++ b/multi/multi_schema-evolution.html @@ -1,6 +1,6 @@ - 10. Schema Evolution Support
10. Schema Evolution Support
Prev Part II. Reference Guide Next

10. Schema Evolution Support

Spring Cloud Stream provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. + 10. Schema Evolution Support

10. Schema Evolution Support
Prev Part I. Preface Next

10. Schema Evolution Support

Spring Cloud Stream provides support for schema evolution so that the data can be evolved over time and still work with older or newer producers and consumers and vice versa. 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. @@ -23,7 +23,7 @@ A schema is referenceable as a tuple consisting of:

DefaultSchemaRegistryClient that does not cache responses. If you intend to change the default behavior, you can use the client directly on your code and override it to the desired outcome. To do so, you have to add the property spring.cloud.stream.schemaRegistryClient.cached=true to your application properties.

10.1.1 Schema Registry Client Properties

The Schema Registry Client supports the following properties:

spring.cloud.stream.schemaRegistryClient.endpoint
The location of the schema-server. -When setting this, use a full URL, including protocol (http or https) , port, and context path.
Default
localhost:8990/
spring.cloud.stream.schemaRegistryClient.cached
Whether the client should cache schema server responses. +When setting this, use a full URL, including protocol (http or https) , port, and context path.
Default
http://localhost:8990/
spring.cloud.stream.schemaRegistryClient.cached
Whether the client should cache schema server responses. Normally set to false, as the caching happens in the message converter. Clients using the schema registry client should set this to true.
Default
false

10.2 Avro Schema Registry Client Message Converters

For applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream auto configures an Apache Avro message converter 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, if the content type of the channel is set to application/*+avro, the MessageConverter is activated, as shown in the following example:

spring.cloud.stream.bindings.output.contentType=application/*+avro

During the outbound conversion, the message converter tries to infer the schema of each outbound messages (based on its type) and register it to a subject (based on the payload type) by using the SchemaRegistryClient. @@ -79,10 +79,10 @@ If you want to use the Confluent schema registry, you need to create a bean of t return client; }

[Note]Note

The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0.

10.6 Schema Registration and Resolution

To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections:

10.6.1 Schema Registration Process (Serialization)

The first part of the registration process is extracting a schema from the payload that is being sent over a channel. Avro types such as SpecificRecord or GenericRecord already contain a schema, which can be retrieved immediately from the instance. -In the case of POJOs, a schema is inferred if the spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled property is set to true (the default).

Figure 10.1. Schema Writer Resolution Process

schema resolution

Ones a schema is obtained, the converter loads its metadata (version) from the remote server. +In the case of POJOs, a schema is inferred if the spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled property is set to true (the default).

Figure 10.1. Schema Writer Resolution Process

schema resolution

Ones a schema is obtained, the converter loads its metadata (version) from the remote server. First, it queries a local cache. If no result is found, it submits the data to the server, which replies with versioning information. -The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.

Figure 10.2. Schema Registration Process

registration

With the schema version information, the converter sets the contentType header of the message to carry the version information — for example: application/vnd.user.v1+avro.

10.6.2 Schema Resolution Process (Deserialization)

When reading messages that contain version information (that is, a contentType header with a scheme like the one described under “Section 10.6.1, “Schema Registration Process (Serialization)””), the converter queries the Schema server to fetch the writer schema of the message. -Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro’s schema resolution support, reads it into the reader definition (setting defaults and any missing properties).

Figure 10.3. Schema Reading Resolution Process

schema reading

[Note]Note

You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application). +The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.

Figure 10.2. Schema Registration Process

registration

With the schema version information, the converter sets the contentType header of the message to carry the version information — for example: application/vnd.user.v1+avro.

10.6.2 Schema Resolution Process (Deserialization)

When reading messages that contain version information (that is, a contentType header with a scheme like the one described under “Section 10.6.1, “Schema Registration Process (Serialization)””), the converter queries the Schema server to fetch the writer schema of the message. +Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro’s schema resolution support, reads it into the reader definition (setting defaults and any missing properties).

Figure 10.3. Schema Reading Resolution Process

schema reading

[Note]Note

You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application). We suggest taking a moment to read the Avro terminology and understand the process. Spring Cloud Stream always fetches the writer schema to determine how to read a message. -If you want to get Avro’s schema evolution support working, you need to make sure that a readerSchema was properly set for your application.

Prev Up Next
9. Content Type Negotiation Home 11. Inter-Application Communication
\ No newline at end of file +If you want to get Avro’s schema evolution support working, you need to make sure that a readerSchema was properly set for your application.

Prev Up Next
9. Content Type Negotiation Home 11. Inter-Application Communication
\ No newline at end of file diff --git a/multi/multi_spring-cloud-stream-overview-binders.html b/multi/multi_spring-cloud-stream-overview-binders.html index a8952ae81..132174267 100644 --- a/multi/multi_spring-cloud-stream-overview-binders.html +++ b/multi/multi_spring-cloud-stream-overview-binders.html @@ -1,7 +1,7 @@ - 7. Binders
7. Binders
Prev Part II. Reference Guide Next

7. Binders

Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware. -This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.

7.1 Producers and Consumers

The following image shows the general relationship of producers and consumers:

Figure 7.1. Producers and Consumers

producers consumers

A producer is any component that sends messages to a channel. + 7. Binders

7. Binders
Prev Part I. Preface Next

7. Binders

Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware. +This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.

7.1 Producers and Consumers

The following image shows the general relationship of producers and consumers:

Figure 7.1. Producers and Consumers

producers consumers

A producer is any component that sends messages to a channel. The channel can be bound to an external message broker with 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 sends 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. @@ -64,13 +64,13 @@ The following example shows how to add the dependency for the Web framework:

application will not start due to health check failures.

You must also enable the bindings actuator endpoints by setting the following property: --management.endpoints.web.exposure.include=bindings.

Once those prerequisites are satisfied. you should see the following in the logs when application start:

: Mapped "{[/actuator/bindings/{name}],methods=[POST]. . .
 : Mapped "{[/actuator/bindings],methods=[GET]. . .
 : Mapped "{[/actuator/bindings/{name}],methods=[GET]. . .

To visualize the current bindings, access the following URL: -<host>:<port>/actuator/bindings

Alternative, to see a single binding, access one of the URLs similar to the following: -<host>:<port>/actuator/bindings/myBindingName

You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a state argument as JSON, as shown in the following examples:

curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName

[Note]Note

PAUSED and RESUMED work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs. +http://<host>:<port>/actuator/bindings

Alternative, to see a single binding, access one of the URLs similar to the following: +http://<host>:<port>/actuator/bindings/myBindingName

You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a state argument as JSON, as shown in the following examples:

curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName

[Note]Note

PAUSED and RESUMED work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs. Currently, only Kafka binder supports the PAUSED and RESUMED states.

7.7 Binder Configuration Properties

The following properties are available when customizing binder configurations. These properties exposed via org.springframework.cloud.stream.config.BinderProperties

They must be prefixed with spring.cloud.stream.binders.<configurationName>.

type

The binder type. It typically references one of the binders found on the classpath — in particular, a key in a META-INF/spring.binders file.

By default, it has the same value as the configuration name.

inheritEnvironment

Whether the configuration inherits the environment of the application itself.

Default: true.

environment

Root for a set of properties that can be used to customize the environment of the binder. When this property is set, the context in which the binder is being created is not a child of the application context. This setting allows for complete separation between the binder components and the application components.

Default: empty.

defaultCandidate

Whether the binder configuration is a candidate for being considered a default binder or can be used only when explicitly referenced. -This setting allows adding binder configurations without interfering with the default processing.

Default: true.

Prev Up Next
6. Programming Model Home 8. Configuration Options
\ No newline at end of file +This setting allows adding binder configurations without interfering with the default processing.

Default: true.

Prev Up Next
6. Programming Model Home 8. Configuration Options
\ No newline at end of file diff --git a/multi/multi_spring-cloud-stream-overview-introducing.html b/multi/multi_spring-cloud-stream-overview-introducing.html index 7916f56f6..39651eeae 100644 --- a/multi/multi_spring-cloud-stream-overview-introducing.html +++ b/multi/multi_spring-cloud-stream-overview-introducing.html @@ -1,6 +1,6 @@ - 4. Introducing Spring Cloud Stream
4. Introducing Spring Cloud Stream
Prev Part II. Reference Guide Next

4. Introducing Spring Cloud Stream

Spring Cloud Stream is a framework for building message-driven microservice applications. + 4. Introducing Spring Cloud Stream

4. Introducing Spring Cloud Stream
Prev Part I. Preface Next

4. Introducing Spring Cloud Stream

Spring Cloud Stream is a framework for building message-driven microservice applications. Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions.

You can add the @EnableBinding annotation to your application to get immediate connectivity to a message broker, and you can add @StreamListener to a method to cause it to receive events for stream processing. The following example shows a sink application that receives external messages:

@SpringBootApplication
@@ -39,4 +39,4 @@ You can use this in the application by autowiring it, as shown in the following
   public void contextLoads() {
     assertNotNull(this.sink.input());
   }
-}
Prev Up Next
Part II. Reference Guide Home 5. Main Concepts
\ No newline at end of file +}
Prev Up Next
3. What’s New in 2.0? Home 5. Main Concepts
\ No newline at end of file diff --git a/multi/multi_spring-cloud-stream-overview-metrics-emitter.html b/multi/multi_spring-cloud-stream-overview-metrics-emitter.html index 6ee3f0cbf..d8741ec4e 100644 --- a/multi/multi_spring-cloud-stream-overview-metrics-emitter.html +++ b/multi/multi_spring-cloud-stream-overview-metrics-emitter.html @@ -1,6 +1,6 @@ - 14. Metrics Emitter
14. Metrics Emitter
Prev Part II. Reference Guide Next

14. Metrics Emitter

Spring Boot Actuator provides dependency management and auto-configuration for Micrometer, an application metrics + 14. Metrics Emitter

14. Metrics Emitter
Prev Part I. Preface Next

14. Metrics Emitter

Spring Boot Actuator provides dependency management and auto-configuration for Micrometer, an application metrics facade that supports numerous monitoring systems.

Spring Cloud Stream provides support for emitting any available micrometer-based metrics to a binding destination, allowing for periodic collection of metric data from stream applications without relying on polling individual endpoints.

Metrics Emitter is activated by defining the spring.cloud.stream.bindings.applicationMetrics.destination property, which specifies the name of the binding destination used by the current binder to publish metric messages.

For example:

spring.cloud.stream.bindings.applicationMetrics.destination=myMetricDestination

The preceding example instructs the binder to bind to myMetricDestination (that is, Rabbit exchange, Kafka topic, and others).

The following properties can be used for customizing the emission of metrics:

spring.cloud.stream.metrics.key

The name of the metric being emitted. Should be a unique value per application.

Default: ${spring.application.name:${vcap.application.name:${spring.config.name:application}}}

spring.cloud.stream.metrics.properties

Allows white listing application properties that are added to the metrics payload

Default: null.

spring.cloud.stream.metrics.meter-filter

Pattern to control the 'meters' one wants to capture. @@ -47,4 +47,4 @@ For example, specifying spring.integration.* captur } ] }

[Note]Note

Given that the format of the Metric message has slightly changed after migrating to Micrometer, the published message will also have -a STREAM_CLOUD_STREAM_VERSION header set to 2.x to help distinguish between Metric messages from the older versions of the Spring Cloud Stream.

Prev Up Next
13. Health Indicator Home 15. Samples
\ No newline at end of file +a STREAM_CLOUD_STREAM_VERSION header set to 2.x to help distinguish between Metric messages from the older versions of the Spring Cloud Stream.

Prev Up Next
13. Health Indicator Home 15. Samples
\ No newline at end of file diff --git a/multi/multi_spring-cloud-stream.html b/multi/multi_spring-cloud-stream.html index 1c4557797..e373737be 100644 --- a/multi/multi_spring-cloud-stream.html +++ b/multi/multi_spring-cloud-stream.html @@ -1,3 +1,3 @@ - Spring Cloud Stream Reference Guide
Spring Cloud Stream Reference Guide
   Next

Spring Cloud Stream Reference Guide

Authors

Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Vinicius Carvalho, Gary Russell, Oleg Zhurakousky, Jay Bryant

Table of Contents

I. Preface
1. A Brief History of Spring’s Data Integration Journey
2. Quick Start
2.1. Creating a Sample Application by Using Spring Initializr
2.2. Importing the Project into Your IDE
2.3. Adding a Message Handler, Building, and Running
3. What’s New in 2.0?
3.1. New Features and Components
3.2. Notable Enhancements
3.2.1. Both Actuator and Web Dependencies Are Now Optional
3.2.2. Content-type Negotiation Improvements
3.3. Notable Deprecations
3.3.1. Java Serialization (Java Native and Kryo)
3.3.2. Deprecated Classes and Methods
II. Reference Guide
4. Introducing Spring Cloud Stream
5. Main Concepts
5.1. Application Model
5.1.1. Fat JAR
5.2. The Binder Abstraction
5.3. Persistent Publish-Subscribe Support
5.4. Consumer Groups
5.5. Consumer Types
5.5.1. Durability
5.6. Partitioning Support
6. Programming Model
6.1. Destination Binders
6.2. Destination Bindings
6.3. Producing and Consuming Messages
6.3.1. Spring Integration Support
6.3.2. Using @StreamListener Annotation
6.3.3. Using @StreamListener for Content-based routing
6.3.4. Spring Cloud Function support
Functional Composition
6.3.5. Using Polled Consumers
Overview
Handling Errors
6.4. Error Handling
6.4.1. Application Error Handling
6.4.2. System Error Handling
Drop Failed Messages
DLQ - Dead Letter Queue
Re-queue Failed Messages
6.4.3. Retry Template
6.5. Reactive Programming Support
6.5.1. Reactor-based Handlers
6.5.2. Reactive Sources
7. Binders
7.1. Producers and Consumers
7.2. Binder SPI
7.3. Binder Detection
7.3.1. Classpath Detection
7.4. Multiple Binders on the Classpath
7.5. Connecting to Multiple Systems
7.6. Binding visualization and control
7.7. Binder Configuration Properties
8. Configuration Options
8.1. Binding Service Properties
8.2. Binding Properties
8.2.1. Common Binding Properties
8.2.2. Consumer Properties
8.2.3. Producer Properties
8.3. Using Dynamically Bound Destinations
9. Content Type Negotiation
9.1. Mechanics
9.1.1. Content Type versus Argument Type
9.1.2. Message Converters
9.2. Provided MessageConverters
9.3. User-defined Message Converters
10. Schema Evolution Support
10.1. Schema Registry Client
10.1.1. Schema Registry Client Properties
10.2. Avro Schema Registry Client Message Converters
10.2.1. Avro Schema Registry Message Converter Properties
10.3. Apache Avro Message Converters
10.4. Converters with Schema Support
10.5. Schema Registry Server
10.5.1. Schema Registry Server API
Registering a New Schema
Retrieving an Existing Schema by Subject, Format, and Version
Retrieving an Existing Schema by Subject and Format
Retrieving an Existing Schema by ID
Deleting a Schema by Subject, Format, and Version
Deleting a Schema by ID
Deleting a Schema by Subject
10.5.2. Using Confluent’s Schema Registry
10.6. Schema Registration and Resolution
10.6.1. Schema Registration Process (Serialization)
10.6.2. Schema Resolution Process (Deserialization)
11. Inter-Application Communication
11.1. Connecting Multiple Application Instances
11.2. Instance Index and Instance Count
11.3. Partitioning
11.3.1. Configuring Output Bindings for Partitioning
11.3.2. Configuring Input Bindings for Partitioning
12. Testing
12.1. Disabling the Test Binder Autoconfiguration
13. Health Indicator
14. Metrics Emitter
15. Samples
15.1. Deploying Stream Applications on CloudFoundry
III. Appendices
A. Building
A.1. Basic Compile and Test
A.2. Documentation
A.3. Working with the code
A.3.1. Importing into eclipse with m2eclipse
A.3.2. Importing into eclipse without m2eclipse
A.4. Sign the Contributor License Agreement
A.5. Code Conventions and Housekeeping
   Next
   Part I. Preface
\ No newline at end of file + Spring Cloud Stream Reference Guide
Spring Cloud Stream Reference Guide
   Next

Spring Cloud Stream Reference Guide

Authors

Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Vinicius Carvalho, Gary Russell, Oleg Zhurakousky, Jay Bryant

Table of Contents

I. Preface
1. A Brief History of Spring’s Data Integration Journey
2. Quick Start
2.1. Creating a Sample Application by Using Spring Initializr
2.2. Importing the Project into Your IDE
2.3. Adding a Message Handler, Building, and Running
3. What’s New in 2.0?
3.1. New Features and Components
3.2. Notable Enhancements
3.2.1. Both Actuator and Web Dependencies Are Now Optional
3.2.2. Content-type Negotiation Improvements
3.3. Notable Deprecations
3.3.1. Java Serialization (Java Native and Kryo)
3.3.2. Deprecated Classes and Methods
4. Introducing Spring Cloud Stream
5. Main Concepts
5.1. Application Model
5.1.1. Fat JAR
5.2. The Binder Abstraction
5.3. Persistent Publish-Subscribe Support
5.4. Consumer Groups
5.5. Consumer Types
5.5.1. Durability
5.6. Partitioning Support
6. Programming Model
6.1. Destination Binders
6.2. Destination Bindings
6.3. Producing and Consuming Messages
6.3.1. Spring Integration Support
6.3.2. Using @StreamListener Annotation
6.3.3. Using @StreamListener for Content-based routing
6.3.4. Spring Cloud Function support
Functional Composition
6.3.5. Using Polled Consumers
Overview
Handling Errors
6.4. Error Handling
6.4.1. Application Error Handling
6.4.2. System Error Handling
Drop Failed Messages
DLQ - Dead Letter Queue
Re-queue Failed Messages
6.4.3. Retry Template
6.5. Reactive Programming Support
6.5.1. Reactor-based Handlers
6.5.2. Reactive Sources
7. Binders
7.1. Producers and Consumers
7.2. Binder SPI
7.3. Binder Detection
7.3.1. Classpath Detection
7.4. Multiple Binders on the Classpath
7.5. Connecting to Multiple Systems
7.6. Binding visualization and control
7.7. Binder Configuration Properties
8. Configuration Options
8.1. Binding Service Properties
8.2. Binding Properties
8.2.1. Common Binding Properties
8.2.2. Consumer Properties
8.2.3. Producer Properties
8.3. Using Dynamically Bound Destinations
9. Content Type Negotiation
9.1. Mechanics
9.1.1. Content Type versus Argument Type
9.1.2. Message Converters
9.2. Provided MessageConverters
9.3. User-defined Message Converters
10. Schema Evolution Support
10.1. Schema Registry Client
10.1.1. Schema Registry Client Properties
10.2. Avro Schema Registry Client Message Converters
10.2.1. Avro Schema Registry Message Converter Properties
10.3. Apache Avro Message Converters
10.4. Converters with Schema Support
10.5. Schema Registry Server
10.5.1. Schema Registry Server API
Registering a New Schema
Retrieving an Existing Schema by Subject, Format, and Version
Retrieving an Existing Schema by Subject and Format
Retrieving an Existing Schema by ID
Deleting a Schema by Subject, Format, and Version
Deleting a Schema by ID
Deleting a Schema by Subject
10.5.2. Using Confluent’s Schema Registry
10.6. Schema Registration and Resolution
10.6.1. Schema Registration Process (Serialization)
10.6.2. Schema Resolution Process (Deserialization)
11. Inter-Application Communication
11.1. Connecting Multiple Application Instances
11.2. Instance Index and Instance Count
11.3. Partitioning
11.3.1. Configuring Output Bindings for Partitioning
11.3.2. Configuring Input Bindings for Partitioning
12. Testing
12.1. Disabling the Test Binder Autoconfiguration
13. Health Indicator
14. Metrics Emitter
15. Samples
15.1. Deploying Stream Applications on CloudFoundry
   Next
   Part I. Preface
\ No newline at end of file diff --git a/single/spring-cloud-stream.html b/single/spring-cloud-stream.html index 6ca2febc9..87da5012f 100644 --- a/single/spring-cloud-stream.html +++ b/single/spring-cloud-stream.html @@ -1,6 +1,6 @@ - Spring Cloud Stream Reference Guide

Spring Cloud Stream Reference Guide

Authors

Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Vinicius Carvalho, Gary Russell, Oleg Zhurakousky, Jay Bryant

Table of Contents

I. Preface
1. A Brief History of Spring’s Data Integration Journey
2. Quick Start
2.1. Creating a Sample Application by Using Spring Initializr
2.2. Importing the Project into Your IDE
2.3. Adding a Message Handler, Building, and Running
3. What’s New in 2.0?
3.1. New Features and Components
3.2. Notable Enhancements
3.2.1. Both Actuator and Web Dependencies Are Now Optional
3.2.2. Content-type Negotiation Improvements
3.3. Notable Deprecations
3.3.1. Java Serialization (Java Native and Kryo)
3.3.2. Deprecated Classes and Methods
II. Reference Guide
4. Introducing Spring Cloud Stream
5. Main Concepts
5.1. Application Model
5.1.1. Fat JAR
5.2. The Binder Abstraction
5.3. Persistent Publish-Subscribe Support
5.4. Consumer Groups
5.5. Consumer Types
5.5.1. Durability
5.6. Partitioning Support
6. Programming Model
6.1. Destination Binders
6.2. Destination Bindings
6.3. Producing and Consuming Messages
6.3.1. Spring Integration Support
6.3.2. Using @StreamListener Annotation
6.3.3. Using @StreamListener for Content-based routing
6.3.4. Spring Cloud Function support
Functional Composition
6.3.5. Using Polled Consumers
Overview
Handling Errors
6.4. Error Handling
6.4.1. Application Error Handling
6.4.2. System Error Handling
Drop Failed Messages
DLQ - Dead Letter Queue
Re-queue Failed Messages
6.4.3. Retry Template
6.5. Reactive Programming Support
6.5.1. Reactor-based Handlers
6.5.2. Reactive Sources
7. Binders
7.1. Producers and Consumers
7.2. Binder SPI
7.3. Binder Detection
7.3.1. Classpath Detection
7.4. Multiple Binders on the Classpath
7.5. Connecting to Multiple Systems
7.6. Binding visualization and control
7.7. Binder Configuration Properties
8. Configuration Options
8.1. Binding Service Properties
8.2. Binding Properties
8.2.1. Common Binding Properties
8.2.2. Consumer Properties
8.2.3. Producer Properties
8.3. Using Dynamically Bound Destinations
9. Content Type Negotiation
9.1. Mechanics
9.1.1. Content Type versus Argument Type
9.1.2. Message Converters
9.2. Provided MessageConverters
9.3. User-defined Message Converters
10. Schema Evolution Support
10.1. Schema Registry Client
10.1.1. Schema Registry Client Properties
10.2. Avro Schema Registry Client Message Converters
10.2.1. Avro Schema Registry Message Converter Properties
10.3. Apache Avro Message Converters
10.4. Converters with Schema Support
10.5. Schema Registry Server
10.5.1. Schema Registry Server API
Registering a New Schema
Retrieving an Existing Schema by Subject, Format, and Version
Retrieving an Existing Schema by Subject and Format
Retrieving an Existing Schema by ID
Deleting a Schema by Subject, Format, and Version
Deleting a Schema by ID
Deleting a Schema by Subject
10.5.2. Using Confluent’s Schema Registry
10.6. Schema Registration and Resolution
10.6.1. Schema Registration Process (Serialization)
10.6.2. Schema Resolution Process (Deserialization)
11. Inter-Application Communication
11.1. Connecting Multiple Application Instances
11.2. Instance Index and Instance Count
11.3. Partitioning
11.3.1. Configuring Output Bindings for Partitioning
11.3.2. Configuring Input Bindings for Partitioning
12. Testing
12.1. Disabling the Test Binder Autoconfiguration
13. Health Indicator
14. Metrics Emitter
15. Samples
15.1. Deploying Stream Applications on CloudFoundry
III. Appendices
A. Building
A.1. Basic Compile and Test
A.2. Documentation
A.3. Working with the code
A.3.1. Importing into eclipse with m2eclipse
A.3.2. Importing into eclipse without m2eclipse
A.4. Sign the Contributor License Agreement
A.5. Code Conventions and Housekeeping

Part I. Preface

1. A Brief History of Spring’s Data Integration Journey

Spring’s journey on Data Integration started with Spring Integration. With its programming model, it provided a consistent developer experience to build applications that can embrace Enterprise Integration Patterns to connect with external systems such as, databases, message brokers, and among others.

Fast forward to the cloud-era, where microservices have become prominent in the enterprise setting. Spring Boot transformed the way how developers built Applications. With Spring’s programming model and the runtime responsibilities handled by Spring Boot, it became seamless to develop stand-alone, production-grade Spring-based microservices.

To extend this to Data Integration workloads, Spring Integration and Spring Boot were put together into a new project. Spring Cloud Stream was born.

With Spring Cloud Stream, developers can: + Spring Cloud Stream Reference Guide

Spring Cloud Stream Reference Guide

Authors

Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Vinicius Carvalho, Gary Russell, Oleg Zhurakousky, Jay Bryant

Table of Contents

I. Preface
1. A Brief History of Spring’s Data Integration Journey
2. Quick Start
2.1. Creating a Sample Application by Using Spring Initializr
2.2. Importing the Project into Your IDE
2.3. Adding a Message Handler, Building, and Running
3. What’s New in 2.0?
3.1. New Features and Components
3.2. Notable Enhancements
3.2.1. Both Actuator and Web Dependencies Are Now Optional
3.2.2. Content-type Negotiation Improvements
3.3. Notable Deprecations
3.3.1. Java Serialization (Java Native and Kryo)
3.3.2. Deprecated Classes and Methods
4. Introducing Spring Cloud Stream
5. Main Concepts
5.1. Application Model
5.1.1. Fat JAR
5.2. The Binder Abstraction
5.3. Persistent Publish-Subscribe Support
5.4. Consumer Groups
5.5. Consumer Types
5.5.1. Durability
5.6. Partitioning Support
6. Programming Model
6.1. Destination Binders
6.2. Destination Bindings
6.3. Producing and Consuming Messages
6.3.1. Spring Integration Support
6.3.2. Using @StreamListener Annotation
6.3.3. Using @StreamListener for Content-based routing
6.3.4. Spring Cloud Function support
Functional Composition
6.3.5. Using Polled Consumers
Overview
Handling Errors
6.4. Error Handling
6.4.1. Application Error Handling
6.4.2. System Error Handling
Drop Failed Messages
DLQ - Dead Letter Queue
Re-queue Failed Messages
6.4.3. Retry Template
6.5. Reactive Programming Support
6.5.1. Reactor-based Handlers
6.5.2. Reactive Sources
7. Binders
7.1. Producers and Consumers
7.2. Binder SPI
7.3. Binder Detection
7.3.1. Classpath Detection
7.4. Multiple Binders on the Classpath
7.5. Connecting to Multiple Systems
7.6. Binding visualization and control
7.7. Binder Configuration Properties
8. Configuration Options
8.1. Binding Service Properties
8.2. Binding Properties
8.2.1. Common Binding Properties
8.2.2. Consumer Properties
8.2.3. Producer Properties
8.3. Using Dynamically Bound Destinations
9. Content Type Negotiation
9.1. Mechanics
9.1.1. Content Type versus Argument Type
9.1.2. Message Converters
9.2. Provided MessageConverters
9.3. User-defined Message Converters
10. Schema Evolution Support
10.1. Schema Registry Client
10.1.1. Schema Registry Client Properties
10.2. Avro Schema Registry Client Message Converters
10.2.1. Avro Schema Registry Message Converter Properties
10.3. Apache Avro Message Converters
10.4. Converters with Schema Support
10.5. Schema Registry Server
10.5.1. Schema Registry Server API
Registering a New Schema
Retrieving an Existing Schema by Subject, Format, and Version
Retrieving an Existing Schema by Subject and Format
Retrieving an Existing Schema by ID
Deleting a Schema by Subject, Format, and Version
Deleting a Schema by ID
Deleting a Schema by Subject
10.5.2. Using Confluent’s Schema Registry
10.6. Schema Registration and Resolution
10.6.1. Schema Registration Process (Serialization)
10.6.2. Schema Resolution Process (Deserialization)
11. Inter-Application Communication
11.1. Connecting Multiple Application Instances
11.2. Instance Index and Instance Count
11.3. Partitioning
11.3.1. Configuring Output Bindings for Partitioning
11.3.2. Configuring Input Bindings for Partitioning
12. Testing
12.1. Disabling the Test Binder Autoconfiguration
13. Health Indicator
14. Metrics Emitter
15. Samples
15.1. Deploying Stream Applications on CloudFoundry

Part I. Preface

1. A Brief History of Spring’s Data Integration Journey

Spring’s journey on Data Integration started with Spring Integration. With its programming model, it provided a consistent developer experience to build applications that can embrace Enterprise Integration Patterns to connect with external systems such as, databases, message brokers, and among others.

Fast forward to the cloud-era, where microservices have become prominent in the enterprise setting. Spring Boot transformed the way how developers built Applications. With Spring’s programming model and the runtime responsibilities handled by Spring Boot, it became seamless to develop stand-alone, production-grade Spring-based microservices.

To extend this to Data Integration workloads, Spring Integration and Spring Boot were put together into a new project. Spring Cloud Stream was born.

With Spring Cloud Stream, developers can: * Build, test, iterate, and deploy data-centric applications in isolation. * Apply modern microservices architecture patterns, including composition through messaging. * Decouple application responsibilities with event-centric thinking. An event can represent something that has happened in time, to which the downstream consumer applications can react without knowing where it originated or the producer’s identity. @@ -78,15 +78,14 @@ See the “this blog post contains more detail.

  • All message conversion is now handled only by MessageConverter objects.
  • We introduced the @StreamMessageConverter annotation to provide custom MessageConverter objects.
  • We introduced the default Content Type as application/json, which needs to be taken into consideration when migrating 1.3 application or operating in the mixed mode (that is, 1.3 producer → 2.0 consumer).
  • Messages with textual payloads and a contentType of text/…​ or …​/json are no longer converted to Message<String> for cases where the argument type of the provided MessageHandler can not be determined (that is, public void handle(Message<?> message) or public void handle(Object payload)). Furthermore, a strong argument type may not be enough to properly convert messages, so the contentType header may be used as a supplement by some MessageConverters.

3.3 Notable Deprecations

As of version 2.0, the following items have been deprecated:

3.3.1 Java Serialization (Java Native and Kryo)

JavaSerializationMessageConverter and KryoMessageConverter remain for now. However, we plan to move them out of the core packages and support in the future. The main reason for this deprecation is to flag the issue that type-based, language-specific serialization could cause in distributed environments, where Producers and Consumers may depend on different JVM versions or have different versions of supporting libraries (that is, Kryo). -We also wanted to draw the attention to the fact that Consumers and Producers may not even be Java-based, so polyglot style serialization (i.e., JSON) is better suited.

3.3.2 Deprecated Classes and Methods

The following is a quick summary of notable deprecations. See the corresponding javadoc for more details.

  • SharedChannelRegistry. Use SharedBindingTargetRegistry.
  • Bindings. +We also wanted to draw the attention to the fact that Consumers and Producers may not even be Java-based, so polyglot style serialization (i.e., JSON) is better suited.

3.3.2 Deprecated Classes and Methods

The following is a quick summary of notable deprecations. See the corresponding {spring-cloud-stream-javadoc-current}[javadoc] for more details.

  • SharedChannelRegistry. Use SharedBindingTargetRegistry.
  • Bindings. Beans qualified by it are already uniquely identified by their type — for example, provided Source, Processor, or custom bindings:
public interface Sample {
 	String OUTPUT = "sampleOutput";
 
 	@Output(Sample.OUTPUT)
 	MessageChannel output();
 }
  • HeaderMode.raw. Use none, headers or embeddedHeaders
  • ProducerProperties.partitionKeyExtractorClass in favor of partitionKeyExtractorName and ProducerProperties.partitionSelectorClass in favor of partitionSelectorName. -This change ensures that both components are Spring configured and managed and are referenced in a Spring-friendly way.
  • BinderAwareRouterBeanPostProcessor. While the component remains, it is no longer a BeanPostProcessor and will be renamed in the future.
  • BinderProperties.setEnvironment(Properties environment). Use BinderProperties.setEnvironment(Map<String, Object> environment).

Part II. Reference Guide

This section goes into more detail about how you can work with Spring Cloud Stream. -It covers topics such as creating and running stream applications.

4. Introducing Spring Cloud Stream

Spring Cloud Stream is a framework for building message-driven microservice applications. +This change ensures that both components are Spring configured and managed and are referenced in a Spring-friendly way.

  • BinderAwareRouterBeanPostProcessor. While the component remains, it is no longer a BeanPostProcessor and will be renamed in the future.
  • BinderProperties.setEnvironment(Properties environment). Use BinderProperties.setEnvironment(Map<String, Object> environment).

    • 4. Introducing Spring Cloud Stream

    Spring Cloud Stream is a framework for building message-driven microservice applications. Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions.

    You can add the @EnableBinding annotation to your application to get immediate connectivity to a message broker, and you can add @StreamListener to a method to cause it to receive events for stream processing. The following example shows a sink application that receives external messages:

    @SpringBootApplication
@@ -128,7 +127,7 @@ You can use this in the application by autowiring it, as shown in the following
 }

    5. Main Concepts

    Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications. This section gives an overview of the following:

    5.1 Application Model

    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.

    Figure 5.1. Spring Cloud Stream Application

    SCSt with binder

    5.1.1 Fat JAR

    Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing. +Channels are connected to external brokers through middleware-specific Binder implementations.

    Figure 5.1. Spring Cloud Stream Application

    SCSt with binder

    5.1.1 Fat JAR

    Spring Cloud Stream applications can be run in stand-alone mode from your IDE for testing. To run a Spring Cloud Stream application in production, you can create an executable (or “fat”) JAR by using the standard Spring Boot tooling provided for Maven or Gradle. See the Spring Boot Reference Guide for more details.

    5.2 The Binder Abstraction

    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 also 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. @@ -138,7 +137,7 @@ In the sink example from the

    5.3 Persistent Publish-Subscribe Support

    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.

    Figure 5.2. Spring Cloud Stream Publish-Subscribe

    SCSt sensors

    Data reported by sensors to an HTTP endpoint is sent to a common destination named raw-sensor-data. +This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.

    Figure 5.2. Spring Cloud Stream Publish-Subscribe

    SCSt sensors

    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 (Hadoop Distributed File System). 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 lets new applications 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. @@ -148,7 +147,7 @@ By using native middleware support, Spring Cloud Stream also simplifies use of t When doing so, 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.

    Figure 5.3. Spring Cloud Stream Consumer Groups

    SCSt groups

    All groups that 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. +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.

    Figure 5.3. Spring Cloud Stream Consumer Groups

    SCSt groups

    All groups that 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.

    5.5 Consumer Types

    Two types of consumer are supported:

    • Message-driven (sometimes referred to as Asynchronous)
    • Polled (sometimes referred to as Synchronous)

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

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

    5.5.1 Durability

    Consistent with the opinionated application model of Spring Cloud Stream, consumer group subscriptions are durable. That is, a binder implementation ensures that group subscriptions are persistent and that, once at least one subscription for a group has been created, the group receives messages, even if they are sent while all applications in the group are stopped.

    [Note]Note

    Anonymous subscriptions are non-durable by nature. For some binder implementations (such as RabbitMQ), it is possible to have non-durable group subscriptions.

    In general, it is preferable to always specify a consumer group when binding an application to a given destination. @@ -156,7 +155,7 @@ When scaling up a Spring Cloud Stream application, you must specify a consumer g Doing so prevents the application’s instances from receiving duplicate messages (unless that behavior is desired, which is unusual).

    5.6 Partitioning Support

    Spring Cloud Stream provides support for partitioning data between multiple instances of a given application. In a partitioned scenario, the physical communication medium (such as 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 (for example, Kafka) or not (for example, RabbitMQ).

    Figure 5.4. Spring Cloud Stream Partitioning

    SCSt partitioning

    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. +Partitioning can thus be used whether the broker itself is naturally partitioned (for example, Kafka) or not (for example, RabbitMQ).

    Figure 5.4. Spring Cloud Stream Partitioning

    SCSt partitioning

    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.

    [Note]Note

    To set up a partitioned processing scenario, you must configure both the data-producing and the data-consuming ends.

    6. Programming Model

    To understand the programming model, you should be familiar with the following core concepts:

    • Destination Binders: Components responsible to provide integration with the external messaging systems.
    • Destination Bindings: Bridge between the external messaging systems and application provided Producers and Consumers of messages (created by the Destination Binders).
    • Message: The canonical data structure used by producers and consumers to communicate with Destination Binders (and thus other applications via external messaging systems).
    SCSt overview

    6.1 Destination Binders

    Destination Binders are extension components of Spring Cloud Stream responsible for providing the necessary configuration and implementation to facilitate integration with external messaging systems. This integration is responsible for connectivity, delegation, and routing of messages to and from producers and consumers, data type conversion, @@ -407,7 +406,7 @@ If the service activator throws a RequeueCurrentMessageExc The error handling comes in two flavors:

    • application: The error handling is done within the application (custom error handler).
    • system: The error handling is delegated to the binder (re-queue, DL, and others). Note that the techniques are dependent on binder implementation and the capability of the underlying messaging middleware.

    Spring Cloud Stream uses the Spring Retry library to facilitate successful message processing. See Section 6.4.3, “Retry Template” for more details. However, when all fails, the exceptions thrown by the message handlers are propagated back to the binder. At that point, binder invokes custom error handler or communicates -the error back to the messaging system (re-queue, DLQ, and others).

    6.4.1 Application Error Handling

    There are two types of application-level error handling. Errors can be handled at each binding subscription or a global handler can handle all the binding subscription errors. Let’s review the details.

    Figure 6.1. A Spring Cloud Stream Sink Application with Custom and Global Error Handlers

    custom vs global error channels

    For each input binding, Spring Cloud Stream creates a dedicated error channel with the following semantics <destinationName>.errors.

    [Note]Note

    The <destinationName> consists of the name of the binding (such as input) and the name of the group (such as myGroup).

    Consider the following:

    spring.cloud.stream.bindings.input.group=myGroup
    @StreamListener(Sink.INPUT) // destination name 'input.myGroup'
+the error back to the messaging system (re-queue, DLQ, and others).
    6.4.1 Application Error Handling
    There are two types of application-level error handling. Errors can be handled at each binding subscription or a global handler can handle all the binding subscription errors. Let’s review the details.
    Figure 6.1. A Spring Cloud Stream Sink Application with Custom and Global Error Handlers
    custom vs global error channels

    For each input binding, Spring Cloud Stream creates a dedicated error channel with the following semantics <destinationName>.errors.
    [Note]Note
    The <destinationName> consists of the name of the binding (such as input) and the name of the group (such as myGroup).
    Consider the following:
    spring.cloud.stream.bindings.input.group=myGroup
    @StreamListener(Sink.INPUT) // destination name 'input.myGroup'
 public void handle(Person value) {
 	throw new RuntimeException("BOOM!");
 }
@@ -550,7 +549,7 @@ The Publisher in the following example still uses R
         .toReactivePublisher();
   }
 }

    7. Binders

    Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware. -This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.

    7.1 Producers and Consumers

    The following image shows the general relationship of producers and consumers:

    Figure 7.1. Producers and Consumers

    producers consumers

    A producer is any component that sends messages to a channel. +This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.

    7.1 Producers and Consumers

    The following image shows the general relationship of producers and consumers:

    Figure 7.1. Producers and Consumers

    producers consumers

    A producer is any component that sends messages to a channel. The channel can be bound to an external message broker with 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 sends 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. @@ -613,11 +612,11 @@ The following example shows how to add the dependency for the Web framework:

    application will not start due to health check failures.

    You must also enable the bindings actuator endpoints by setting the following property: --management.endpoints.web.exposure.include=bindings.

    Once those prerequisites are satisfied. you should see the following in the logs when application start:

    : Mapped "{[/actuator/bindings/{name}],methods=[POST]. . .
 : Mapped "{[/actuator/bindings],methods=[GET]. . .
 : Mapped "{[/actuator/bindings/{name}],methods=[GET]. . .

    To visualize the current bindings, access the following URL: -<host>:<port>/actuator/bindings

    Alternative, to see a single binding, access one of the URLs similar to the following: -<host>:<port>/actuator/bindings/myBindingName

    You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a state argument as JSON, as shown in the following examples:

    curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName

    [Note]Note

    PAUSED and RESUMED work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs. +http://<host>:<port>/actuator/bindings

    Alternative, to see a single binding, access one of the URLs similar to the following: +http://<host>:<port>/actuator/bindings/myBindingName

    You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a state argument as JSON, as shown in the following examples:

    curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName

    [Note]Note

    PAUSED and RESUMED work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs. Currently, only Kafka binder supports the PAUSED and RESUMED states.

    7.7 Binder Configuration Properties

    The following properties are available when customizing binder configurations. These properties exposed via org.springframework.cloud.stream.config.BinderProperties

    They must be prefixed with spring.cloud.stream.binders.<configurationName>.

    type

    The binder type. It typically references one of the binders found on the classpath — in particular, a key in a META-INF/spring.binders file.

    By default, it has the same value as the configuration name.

    inheritEnvironment

    Whether the configuration inherits the environment of the application itself.

    Default: true.

    environment

    Root for a set of properties that can be used to customize the environment of the binder. When this property is set, the context in which the binder is being created is not a child of the application context. @@ -844,7 +843,7 @@ A schema is referenceable as a tuple consisting of:

    DefaultSchemaRegistryClient that does not cache responses. If you intend to change the default behavior, you can use the client directly on your code and override it to the desired outcome. To do so, you have to add the property spring.cloud.stream.schemaRegistryClient.cached=true to your application properties.

    10.1.1 Schema Registry Client Properties

    The Schema Registry Client supports the following properties:

    spring.cloud.stream.schemaRegistryClient.endpoint
    The location of the schema-server. -When setting this, use a full URL, including protocol (http or https) , port, and context path.
    Default
    localhost:8990/
    spring.cloud.stream.schemaRegistryClient.cached
    Whether the client should cache schema server responses. +When setting this, use a full URL, including protocol (http or https) , port, and context path.
    Default
    http://localhost:8990/
    spring.cloud.stream.schemaRegistryClient.cached
    Whether the client should cache schema server responses. Normally set to false, as the caching happens in the message converter. Clients using the schema registry client should set this to true.
    Default
    false

    10.2 Avro Schema Registry Client Message Converters

    For applications that have a SchemaRegistryClient bean registered with the application context, Spring Cloud Stream auto configures an Apache Avro message converter 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, if the content type of the channel is set to application/*+avro, the MessageConverter is activated, as shown in the following example:

    spring.cloud.stream.bindings.output.contentType=application/*+avro

    During the outbound conversion, the message converter tries to infer the schema of each outbound messages (based on its type) and register it to a subject (based on the payload type) by using the SchemaRegistryClient. @@ -900,10 +899,10 @@ If you want to use the Confluent schema registry, you need to create a bean of t return client; }

    [Note]Note

    The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0.

    10.6 Schema Registration and Resolution

    To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections:

    10.6.1 Schema Registration Process (Serialization)

    The first part of the registration process is extracting a schema from the payload that is being sent over a channel. Avro types such as SpecificRecord or GenericRecord already contain a schema, which can be retrieved immediately from the instance. -In the case of POJOs, a schema is inferred if the spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled property is set to true (the default).

    Figure 10.1. Schema Writer Resolution Process

    schema resolution

    Ones a schema is obtained, the converter loads its metadata (version) from the remote server. +In the case of POJOs, a schema is inferred if the spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled property is set to true (the default).

    Figure 10.1. Schema Writer Resolution Process

    schema resolution

    Ones a schema is obtained, the converter loads its metadata (version) from the remote server. First, it queries a local cache. If no result is found, it submits the data to the server, which replies with versioning information. -The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.

    Figure 10.2. Schema Registration Process

    registration

    With the schema version information, the converter sets the contentType header of the message to carry the version information — for example: application/vnd.user.v1+avro.

    10.6.2 Schema Resolution Process (Deserialization)

    When reading messages that contain version information (that is, a contentType header with a scheme like the one described under “Section 10.6.1, “Schema Registration Process (Serialization)””), the converter queries the Schema server to fetch the writer schema of the message. -Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro’s schema resolution support, reads it into the reader definition (setting defaults and any missing properties).

    Figure 10.3. Schema Reading Resolution Process

    schema reading

    [Note]Note

    You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application). +The converter always caches the results to avoid the overhead of querying the Schema Server for every new message that needs to be serialized.

    Figure 10.2. Schema Registration Process

    registration

    With the schema version information, the converter sets the contentType header of the message to carry the version information — for example: application/vnd.user.v1+avro.

    10.6.2 Schema Resolution Process (Deserialization)

    When reading messages that contain version information (that is, a contentType header with a scheme like the one described under “Section 10.6.1, “Schema Registration Process (Serialization)””), the converter queries the Schema server to fetch the writer schema of the message. +Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro’s schema resolution support, reads it into the reader definition (setting defaults and any missing properties).

    Figure 10.3. Schema Reading Resolution Process

    schema reading

    [Note]Note

    You should understand the difference between a writer schema (the application that wrote the message) and a reader schema (the receiving application). We suggest taking a moment to read the Avro terminology and understand the process. Spring Cloud Stream always fetches the writer schema to determine how to read a message. If you want to get Avro’s schema evolution support working, you need to make sure that a readerSchema was properly set for your application.

    11. Inter-Application Communication

    Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:

    11.1 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. @@ -1049,62 +1048,4 @@ For example, specifying spring.integration.* captur } ] }

    [Note]Note

    Given that the format of the Metric message has slightly changed after migrating to Micrometer, the published message will also have -a STREAM_CLOUD_STREAM_VERSION header set to 2.x to help distinguish between Metric messages from the older versions of the Spring Cloud Stream.

    15. Samples

    For Spring Cloud Stream samples, see the spring-cloud-stream-samples repository on GitHub.

    15.1 Deploying Stream Applications on CloudFoundry

    On CloudFoundry, services are usually exposed through 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 Cloud Foundry Server docs.

    Part III. Appendices

    Appendix A. Building

    A.1 Basic Compile and Test

    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

    $ ./mvnw clean install

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

    [Note]Note

    You can also install Maven (>=3.3.3) yourself and run the mvn command -in place of ./mvnw in the examples below. If you do that you also -might need to add -P spring if your local Maven settings do not -contain repository declarations for spring pre-release artifacts.

    [Note]Note

    Be aware that you might need to increase the amount of memory -available to Maven by setting a MAVEN_OPTS environment variable with -a value like -Xmx512m -XX:MaxPermSize=128m. We try to cover this in -the .mvn configuration, so if you find you have to do it to make a -build succeed, please raise a ticket to get the settings added to -source control.

    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.

    A.2 Documentation

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

    A.3 Working with the code

    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.

    A.3.1 Importing into eclipse with m2eclipse

    We recommend the m2eclipe eclipse plugin when working with -eclipse. If you don’t already have m2eclipse installed it is available from the "eclipse -marketplace".

    Unfortunately m2e does not yet support Maven 3.3, so once the projects -are imported into Eclipse you will also need to tell m2eclipse to use -the .settings.xml file for the projects. If you do not do this you -may see many different errors related to the POMs in the -projects. Open your Eclipse preferences, expand the Maven -preferences, and select User Settings. In the User Settings field -click Browse and navigate to the Spring Cloud project you imported -selecting the .settings.xml file in that project. Click Apply and -then OK to save the preference changes.

    [Note]Note

    Alternatively you can copy the repository settings from .settings.xml into your own ~/.m2/settings.xml.

    A.3.2 Importing into eclipse without m2eclipse

    If you prefer not to use m2eclipse you can generate eclipse project metadata using the -following command:

    $ ./mvnw eclipse:eclipse

    The generated eclipse projects can be imported by selecting import existing projects -from the file menu. -[[contributing] -== Contributing

    Spring Cloud is released under the non-restrictive Apache 2.0 license, -and follows a very standard Github development process, using Github -tracker for issues and merging pull requests into master. If you want -to contribute even something trivial please do not hesitate, but -follow the guidelines below.

    A.4 Sign the Contributor License Agreement

    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.

    A.5 Code Conventions and Housekeeping

    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.

    • Use the Spring Framework code format conventions. If you use Eclipse -you can import formatter settings using the -eclipse-code-formatter.xml file from the -Spring -Cloud Build project. If using IntelliJ, you can use the -Eclipse Code Formatter -Plugin to import the same file.
    • Make sure all new .java files to have a simple Javadoc class comment with at least an -@author tag identifying you, and preferably at least a paragraph on what the class is -for.
    • Add the ASF license header comment to all new .java files (copy from existing files -in the project)
    • Add yourself as an @author to the .java files that you modify substantially (more -than cosmetic changes).
    • Add some Javadocs and, if you change the namespace, some XSD doc elements.
    • A few unit tests would help a lot as well — someone has to do it.
    • If no-one else is using your branch, please rebase it against the current master (or -other target branch in the main project).
    • When writing a commit message please follow these conventions, -if you are fixing an existing issue please add Fixes gh-XXXX at the end of the commit -message (where XXXX is the issue number).
    \ No newline at end of file +a STREAM_CLOUD_STREAM_VERSION header set to 2.x to help distinguish between Metric messages from the older versions of the Spring Cloud Stream.

    15. Samples

    For Spring Cloud Stream samples, see the spring-cloud-stream-samples repository on GitHub.

    15.1 Deploying Stream Applications on CloudFoundry

    On CloudFoundry, services are usually exposed through 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 Cloud Foundry Server docs.

    \ No newline at end of file diff --git a/spring-cloud-stream.xml b/spring-cloud-stream.xml index dfcff67ed..131f13f59 100644 --- a/spring-cloud-stream.xml +++ b/spring-cloud-stream.xml @@ -1,5 +1,5 @@ - + @@ -376,7 +376,7 @@ We also wanted to draw the attention to the fact that Consumers and Producers ma
    Deprecated Classes and Methods -The following is a quick summary of notable deprecations. See the corresponding javadoc for more details. +The following is a quick summary of notable deprecations. See the corresponding {spring-cloud-stream-javadoc-current}[javadoc] for more details. SharedChannelRegistry. Use SharedBindingTargetRegistry. @@ -407,16 +407,10 @@ This change ensures that both components are Spring configured and managed and a BinderProperties.setEnvironment(Properties environment). Use BinderProperties.setEnvironment(Map<String, Object> environment). +
    - - -Reference Guide - -This section goes into more detail about how you can work with Spring Cloud Stream. -It covers topics such as creating and running stream applications. - Introducing Spring Cloud Stream Spring Cloud Stream is a framework for building message-driven microservice applications. @@ -1533,14 +1527,14 @@ application will not start due to health check failures. : Mapped "{[/actuator/bindings],methods=[GET]. . . : Mapped "{[/actuator/bindings/{name}],methods=[GET]. . . To visualize the current bindings, access the following URL: -<host>:<port>/actuator/bindings +http://<host>:<port>/actuator/bindings Alternative, to see a single binding, access one of the URLs similar to the following: -<host>:<port>/actuator/bindings/myBindingName +http://<host>:<port>/actuator/bindings/myBindingName You can also stop, start, pause, and resume individual bindings by posting to the same URL while providing a state argument as JSON, as shown in the following examples: -curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName -curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST <host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"PAUSED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName +curl -d '{"state":"RESUMED"}' -H "Content-Type: application/json" -X POST http://<host>:<port>/actuator/bindings/myBindingName PAUSED and RESUMED work only when the corresponding binder and its underlying technology supports it. Otherwise, you see the warning message in the logs. Currently, only Kafka binder supports the PAUSED and RESUMED states. @@ -2223,7 +2217,7 @@ When setting this, use a full URL, including protocol (http o Default -localhost:8990/ +http://localhost:8990/ @@ -2876,140 +2870,4 @@ a STREAM_CLOUD_STREAM_VERSION header set to 2.x - -Appendices - -Building -
    -Basic Compile and Test -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 -$ ./mvnw clean install -You can also add '-DskipTests' if you like, to avoid running the tests. - -You can also install Maven (>=3.3.3) yourself and run the mvn command -in place of ./mvnw in the examples below. If you do that you also -might need to add -P spring if your local Maven settings do not -contain repository declarations for spring pre-release artifacts. - - -Be aware that you might need to increase the amount of memory -available to Maven by setting a MAVEN_OPTS environment variable with -a value like -Xmx512m -XX:MaxPermSize=128m. We try to cover this in -the .mvn configuration, so if you find you have to do it to make a -build succeed, please raise a ticket to get the settings added to -source control. - -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. -
    -
    -Documentation -There is a "full" profile that will generate documentation. -
    -
    -Working with the code -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. -
    -Importing into eclipse with m2eclipse -We recommend the m2eclipe eclipse plugin when working with -eclipse. If you don’t already have m2eclipse installed it is available from the "eclipse -marketplace". -Unfortunately m2e does not yet support Maven 3.3, so once the projects -are imported into Eclipse you will also need to tell m2eclipse to use -the .settings.xml file for the projects. If you do not do this you -may see many different errors related to the POMs in the -projects. Open your Eclipse preferences, expand the Maven -preferences, and select User Settings. In the User Settings field -click Browse and navigate to the Spring Cloud project you imported -selecting the .settings.xml file in that project. Click Apply and -then OK to save the preference changes. - -Alternatively you can copy the repository settings from .settings.xml into your own ~/.m2/settings.xml. - -
    -
    -Importing into eclipse without m2eclipse -If you prefer not to use m2eclipse you can generate eclipse project metadata using the -following command: -$ ./mvnw eclipse:eclipse -The generated eclipse projects can be imported by selecting import existing projects -from the file menu. -[[contributing] -== Contributing -Spring Cloud is released under the non-restrictive Apache 2.0 license, -and follows a very standard Github development process, using Github -tracker for issues and merging pull requests into master. If you want -to contribute even something trivial please do not hesitate, but -follow the guidelines below. -
    -
    -
    -Sign the Contributor License Agreement -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. -
    -
    -Code Conventions and Housekeeping -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. - - -Use the Spring Framework code format conventions. If you use Eclipse -you can import formatter settings using the -eclipse-code-formatter.xml file from the -Spring -Cloud Build project. If using IntelliJ, you can use the -Eclipse Code Formatter -Plugin to import the same file. - - -Make sure all new .java files to have a simple Javadoc class comment with at least an -@author tag identifying you, and preferably at least a paragraph on what the class is -for. - - -Add the ASF license header comment to all new .java files (copy from existing files -in the project) - - -Add yourself as an @author to the .java files that you modify substantially (more -than cosmetic changes). - - -Add some Javadocs and, if you change the namespace, some XSD doc elements. - - -A few unit tests would help a lot as well — someone has to do it. - - -If no-one else is using your branch, please rebase it against the current master (or -other target branch in the main project). - - -When writing a commit message please follow these conventions, -if you are fixing an existing issue please add Fixes gh-XXXX at the end of the commit -message (where XXXX is the issue number). - - -
    -     -         \ No newline at end of file