Table of Contents
IRule
Location
headerPropertySourceLocator
behaviorSpring Cloud provides tools for developers to quickly build some of the common patterns in distributed systems (e.g. configuration management, service discovery, circuit breakers, intelligent routing, micro-proxy, control bus). Coordination of distributed systems leads to boiler plate patterns, and using Spring Cloud developers can quickly stand up services and applications that implement those patterns. They will work well in any distributed environment, including the developer’s own laptop, bare metal data centres, and managed platforms such as Cloud Foundry.
Version: 1.0.0.BUILD-SNAPSHOT
Spring Cloud focuses on providing good out of box experience for typical use cases and extensibility mechanism to cover others.
Cloud Native is a style of application development that encourages easy adoption of best practices in the areas of continuous delivery and value-driven development. A related discipline is that of building 12-factor Applications, in which development practices are aligned with delivery and operations goals — for instance, by using declarative programming and management and monitoring. Spring Cloud facilitates these styles of development in a number of specific ways. The starting point is a set of features to which all components in a distributed system need easy access.
Many of those features are covered by Spring Boot, on which Spring Cloud builds. Some more features are delivered by Spring Cloud as two libraries: Spring Cloud Context and Spring Cloud Commons.
Spring Cloud Context provides utilities and special services for the ApplicationContext
of a Spring Cloud application (bootstrap context, encryption, refresh scope, and environment endpoints). Spring Cloud Commons is a set of abstractions and common classes used in different Spring Cloud implementations (such as Spring Cloud Netflix and Spring Cloud Consul).
If you get an exception due to "Illegal key size" and you use Sun’s JDK, you need to install the Java Cryptography Extension (JCE) Unlimited Strength Jurisdiction Policy Files. See the following links for more information:
Extract the files into the JDK/jre/lib/security folder for whichever version of JRE/JDK x64/x86 you use.
![]() | Note |
---|---|
Spring Cloud is released under the non-restrictive Apache 2.0 license. If you would like to contribute to this section of the documentation or if you find an error, you can find the source code and issue trackers for the project at github. |
Spring Boot has an opinionated view of how to build an application with Spring. For instance, it has conventional locations for common configuration files and has endpoints for common management and monitoring tasks. Spring Cloud builds on top of that and adds a few features that probably all components in a system would use or occasionally need.
A Spring Cloud application operates by creating a “bootstrap” context, which is a parent context for the main application.
It is responsible for loading configuration properties from the external sources and for decrypting properties in the local external configuration files.
The two contexts share an Environment
, which is the source of external properties for any Spring application.
By default, bootstrap properties (not bootstrap.properties
but properties that are loaded during the bootstrap phase) are added with high precedence, so they cannot be overridden by local configuration.
The bootstrap context uses a different convention for locating external configuration than the main application context.
Instead of application.yml
(or .properties
), you can use bootstrap.yml
, keeping the external configuration for bootstrap and main context
nicely separate.
The following listing shows an example:
bootstrap.yml.
spring: application: name: foo cloud: config: uri: ${SPRING_CONFIG_URI:http://localhost:8888}
If your application needs any application-specific configuration from the server, it is a good idea to set the spring.application.name
(in bootstrap.yml
or application.yml
).
You can disable the bootstrap process completely by setting spring.cloud.bootstrap.enabled=false
(for example, in system properties).
If you build an application context from SpringApplication
or SpringApplicationBuilder
, then the Bootstrap context is added as a parent to that context.
It is a feature of Spring that child contexts inherit property sources and profiles from their parent, so the “main” application context contains additional property sources, compared to building the same context without Spring Cloud Config.
The additional property sources are:
PropertySourceLocators
are found in the Bootstrap context and if they have non-empty properties, an optional CompositePropertySource
appears with high priority.
An example would be properties from the Spring Cloud Config Server.
See “Section 2.6, “Customizing the Bootstrap Property Sources”” for instructions on how to customize the contents of this property source.bootstrap.yml
(or .properties
), those properties are used to configure the Bootstrap context.
Then they get added to the child context when its parent is set.
They have lower precedence than the application.yml
(or .properties
) and any other property sources that are added to the child as a normal part of the process of creating a Spring Boot application.
See “Section 2.3, “Changing the Location of Bootstrap Properties”” for instructions on how to customize the contents of these property sources.Because of the ordering rules of property sources, the “bootstrap” entries take precedence.
However, note that these do not contain any data from bootstrap.yml
, which has very low precedence but can be used to set defaults.
You can extend the context hierarchy by setting the parent context of any ApplicationContext
you create — for example, by using its own interface or with the SpringApplicationBuilder
convenience methods (parent()
, child()
and sibling()
).
The bootstrap context is the parent of the most senior ancestor that you create yourself.
Every context in the hierarchy has its own “bootstrap” (possibly empty) property source to avoid promoting values inadvertently from parents down to their descendants.
If there is a Config Server, every context in the hierarchy can also (in principle) have a different spring.application.name
and, hence, a different remote property source.
Normal Spring application context behavior rules apply to property resolution: properties from a child context override those in
the parent, by name and also by property source name.
(If the child has a property source with the same name as the parent, the value from the parent is not included in the child).
Note that the SpringApplicationBuilder
lets you share an Environment
amongst the whole hierarchy, but that is not the default.
Thus, sibling contexts, in particular, do not need to have the same profiles or property sources, even though they may share common values with their parent.
The bootstrap.yml
(or .properties
) location can be specified by setting spring.cloud.bootstrap.name
(default: bootstrap
) or spring.cloud.bootstrap.location
(default: empty) — for example, in System properties.
Those properties behave like the spring.config.*
variants with the same name.
In fact, they are used to set up the bootstrap ApplicationContext
by setting those properties in its Environment
.
If there is an active profile (from spring.profiles.active
or through the Environment
API in the
context you are building), properties in that profile get loaded as well, the same as in a regular Spring Boot app — for example, from bootstrap-development.properties
for a development
profile.
The property sources that are added to your application by the bootstrap context are often “remote” (from example, from Spring Cloud Config Server).
By default, they cannot be overridden locally.
If you want to let your applications override the remote properties with their own System properties or config files, the remote property source has to grant it permission by setting spring.cloud.config.allowOverride=true
(it does not work to set this locally).
Once that flag is set, two finer-grained settings control the location of the remote properties in relation to system properties and the application’s local configuration:
spring.cloud.config.overrideNone=true
: Override from any local property source.spring.cloud.config.overrideSystemProperties=false
: Only system properties, command line arguments, and environment variables (but not the local config files) should override the remote settings.The bootstrap context can be set to do anything you like by adding entries to /META-INF/spring.factories
under a key named org.springframework.cloud.bootstrap.BootstrapConfiguration
.
This holds a comma-separated list of Spring @Configuration
classes that are used to create the context.
Any beans that you want to be available to the main application context for autowiring can be created here.
There is a special contract for @Beans
of type ApplicationContextInitializer
.
If you want to control the startup sequence, classes can be marked with an @Order
annotation (the default order is last
).
![]() | Warning |
---|---|
When adding custom |
The bootstrap process ends by injecting initializers into the main SpringApplication
instance (which is the normal Spring Boot startup sequence, whether it is running as a standalone application or deployed in an application server).
First, a bootstrap context is created from the classes found in spring.factories
.
Then, all @Beans
of type ApplicationContextInitializer
are added to the main SpringApplication
before it is started.
The default property source for external configuration added by the bootstrap process is the Spring Cloud Config Server, but you can add additional sources by adding beans of type PropertySourceLocator
to the bootstrap context (through spring.factories
).
For instance, you can insert additional properties from a different server or from a database.
As an example, consider the following custom locator:
@Configuration public class CustomPropertySourceLocator implements PropertySourceLocator { @Override public PropertySource<?> locate(Environment environment) { return new MapPropertySource("customProperty", Collections.<String, Object>singletonMap("property.from.sample.custom.source", "worked as intended")); } }
The Environment
that is passed in is the one for the ApplicationContext
about to be created — in other words, the one for which we supply additional property sources for.
It already has its normal Spring Boot-provided property sources, so you can use those to locate a property source specific to this Environment
(for example, by keying it on spring.application.name
, as is done in the default Spring Cloud Config Server property source locator).
If you create a jar with this class in it and then add a META-INF/spring.factories
containing the following, the customProperty
PropertySource
appears in any application that includes that jar on its classpath:
org.springframework.cloud.bootstrap.BootstrapConfiguration=sample.custom.CustomPropertySourceLocator
If you are going to use Spring Boot to configure log settings than you should place this configuration in `bootstrap.[yml | properties] if you would like it to apply to all events.
![]() | Note |
---|---|
For Spring Cloud to initialize logging configuration properly you cannot use a custom prefix. For example,
using |
The application listens for an EnvironmentChangeEvent
and reacts to the change in a couple of standard ways (additional ApplicationListeners
can be added as @Beans
by the user in the normal way).
When an EnvironmentChangeEvent
is observed, it has a list of key values that have changed, and the application uses those to:
@ConfigurationProperties
beans in the contextlogging.level.*
Note that the Config Client does not, by default, poll for changes in the Environment
.
Generally, we would not recommend that approach for detecting changes (although you could set it up with a
@Scheduled
annotation).
If you have a scaled-out client application, it is better to broadcast the EnvironmentChangeEvent
to all the instances instead of having them polling for changes (for example, by using the Spring Cloud Bus).
The EnvironmentChangeEvent
covers a large class of refresh use cases, as long as you can actually make a change to the Environment
and publish the event.
Note that those APIs are public and part of core Spring).
You can verify that the changes are bound to @ConfigurationProperties
beans by visiting the /configprops
endpoint (a normal Spring Boot Actuator feature).
For instance, a DataSource
can have its maxPoolSize
changed at runtime (the default DataSource
created by Spring Boot is an @ConfigurationProperties
bean) and grow capacity dynamically.
Re-binding @ConfigurationProperties
does not cover another large class of use cases, where you need more control over the refresh and where you need a change to be atomic over the whole ApplicationContext
.
To address those concerns, we have @RefreshScope
.
When there is a configuration change, a Spring @Bean
that is marked as @RefreshScope
gets special treatment.
This feature addresses the problem of stateful beans that only get their configuration injected when they are initialized.
For instance, if a DataSource
has open connections when the database URL is changed via the Environment
, you probably want the holders of those connections to be able to complete what they are doing.
Then, the next time something borrows a connection from the pool, it gets one with the new URL.
Sometimes, it might even be mandatory to apply the @RefreshScope
annotation on some beans which can be only initialized once. If a bean
is "immutable", you will have to either annotate the bean with @RefreshScope
or specify the classname under the property key
spring.cloud.refresh.extra-refreshable
.
![]() | Important |
---|---|
If you create a |
Refresh scope beans are lazy proxies that initialize when they are used (that is, when a method is called), and the scope acts as a cache of initialized values. To force a bean to re-initialize on the next method call, you must invalidate its cache entry.
The RefreshScope
is a bean in the context and has a public refreshAll()
method to refresh all beans in the scope by clearing the target cache.
The /refresh
endpoint exposes this functionality (over HTTP or JMX).
To refresh an individual bean by name, there is also a refresh(String)
method.
To expose the /refresh
endpoint, you need to add following configuration to your application:
management: endpoints: web: exposure: include: refresh
![]() | Note |
---|---|
|
Spring Cloud has an Environment
pre-processor for decrypting property values locally.
It follows the same rules as the Config Server and has the same external configuration through encrypt.*
.
Thus, you can use encrypted values in the form of {cipher}*
and, as long as there is a valid key, they are decrypted before the main application context gets the Environment
settings.
To use the encryption features in an application, you need to include Spring Security RSA in your classpath (Maven co-ordinates: "org.springframework.security:spring-security-rsa"), and you also need the full strength JCE extensions in your JVM.
If you get an exception due to "Illegal key size" and you use Sun’s JDK, you need to install the Java Cryptography Extension (JCE) Unlimited Strength Jurisdiction Policy Files. See the following links for more information:
Extract the files into the JDK/jre/lib/security folder for whichever version of JRE/JDK x64/x86 you use.
For a Spring Boot Actuator application, some additional management endpoints are available. You can use:
POST
to /actuator/env
to update the Environment
and rebind @ConfigurationProperties
and log levels./actuator/refresh
to re-load the boot strap context and refresh the @RefreshScope
beans./actuator/restart
to close the ApplicationContext
and restart it (disabled by default)./actuator/pause
and /actuator/resume
for calling the Lifecycle
methods (stop()
and start()
on the ApplicationContext
).![]() | Note |
---|---|
If you disable the |
Patterns such as service discovery, load balancing, and circuit breakers lend themselves to a common abstraction layer that can be consumed by all Spring Cloud clients, independent of the implementation (for example, discovery with Eureka or Consul).
Spring Cloud Commons provides the @EnableDiscoveryClient
annotation.
This looks for implementations of the DiscoveryClient
interface with META-INF/spring.factories
.
Implementations of the Discovery Client add a configuration class to spring.factories
under the org.springframework.cloud.client.discovery.EnableDiscoveryClient
key.
Examples of DiscoveryClient
implementations include Spring Cloud Netflix Eureka, Spring Cloud Consul Discovery, and Spring Cloud Zookeeper Discovery.
By default, implementations of DiscoveryClient
auto-register the local Spring Boot server with the remote discovery server.
This behavior can be disabled by setting autoRegister=false
in @EnableDiscoveryClient
.
![]() | Note |
---|---|
|
Commons creates a Spring Boot HealthIndicator
that DiscoveryClient
implementations can participate in by implementing DiscoveryHealthIndicator
.
To disable the composite HealthIndicator
, set spring.cloud.discovery.client.composite-indicator.enabled=false
.
A generic HealthIndicator
based on DiscoveryClient
is auto-configured (DiscoveryClientHealthIndicator
).
To disable it, set spring.cloud.discovery.client.health-indicator.enabled=false
.
To disable the description field of the DiscoveryClientHealthIndicator
, set spring.cloud.discovery.client.health-indicator.include-description=false
.
Otherwise, it can bubble up as the description
of the rolled up HealthIndicator
.
DiscoveryClient
interface extends Ordered
. This is useful when using multiple discovery
clients, as it allows you to define the order of the returned discovery clients, similar to
how you can order the beans loaded by a Spring application. By default, the order of any DiscoveryClient
is set to
0
. If you want to set a different order for your custom DiscoveryClient
implementations, you just need to override
the getOrder()
method so that it returns the value that is suitable for your setup. Apart from this, you can use
properties to set the order of the DiscoveryClient
implementations provided by Spring Cloud, among others ConsulDiscoveryClient
, EurekaDiscoveryClient
and
ZookeeperDiscoveryClient
. In order to do it, you just need to set the
spring.cloud.{clientIdentifier}.discovery.order
(or eureka.client.order
for Eureka) property to the desired value.
Commons now provides a ServiceRegistry
interface that provides methods such as register(Registration)
and deregister(Registration)
, which let you provide custom registered services.
Registration
is a marker interface.
The following example shows the ServiceRegistry
in use:
@Configuration @EnableDiscoveryClient(autoRegister=false) public class MyConfiguration { private ServiceRegistry registry; public MyConfiguration(ServiceRegistry registry) { this.registry = registry; } // called through some external process, such as an event or a custom actuator endpoint public void register() { Registration registration = constructRegistration(); this.registry.register(registration); } }
Each ServiceRegistry
implementation has its own Registry
implementation.
ZookeeperRegistration
used with ZookeeperServiceRegistry
EurekaRegistration
used with EurekaServiceRegistry
ConsulRegistration
used with ConsulServiceRegistry
If you are using the ServiceRegistry
interface, you are going to need to pass the
correct Registry
implementation for the ServiceRegistry
implementation you
are using.
By default, the ServiceRegistry
implementation auto-registers the running service.
To disable that behavior, you can set:
* @EnableDiscoveryClient(autoRegister=false)
to permanently disable auto-registration.
* spring.cloud.service-registry.auto-registration.enabled=false
to disable the behavior through configuration.
There are two events that will be fired when a service auto-registers. The first event, called
InstancePreRegisteredEvent
, is fired before the service is registered. The second
event, called InstanceRegisteredEvent
, is fired after the service is registered. You can register an
ApplicationListener
(s) to listen to and react to these events.
![]() | Note |
---|---|
These events will not be fired if |
Spring Cloud Commons provides a /service-registry
actuator endpoint.
This endpoint relies on a Registration
bean in the Spring Application Context.
Calling /service-registry
with GET returns the status of the Registration
.
Using POST to the same endpoint with a JSON body changes the status of the current Registration
to the new value.
The JSON body has to include the status
field with the preferred value.
Please see the documentation of the ServiceRegistry
implementation you use for the allowed values when updating the status and the values returned for the status.
For instance, Eureka’s supported statuses are UP
, DOWN
, OUT_OF_SERVICE
, and UNKNOWN
.
RestTemplate
can be automatically configured to use ribbon.
To create a load-balanced RestTemplate
, create a RestTemplate
@Bean
and use the @LoadBalanced
qualifier, as shown in the following example:
@Configuration public class MyConfiguration { @LoadBalanced @Bean RestTemplate restTemplate() { return new RestTemplate(); } } public class MyClass { @Autowired private RestTemplate restTemplate; public String doOtherStuff() { String results = restTemplate.getForObject("http://stores/stores", String.class); return results; } }
![]() | Caution |
---|---|
A |
The URI needs to use a virtual host name (that is, a service name, not a host name).
The Ribbon client is used to create a full physical address.
See RibbonAutoConfiguration for details of how the RestTemplate
is set up.
WebClient
can be automatically configured to use the LoadBalancerClient
.
To create a load-balanced WebClient
, create a WebClient.Builder
@Bean
and use the @LoadBalanced
qualifier, as shown in the following example:
@Configuration public class MyConfiguration { @Bean @LoadBalanced public WebClient.Builder loadBalancedWebClientBuilder() { return WebClient.builder(); } } public class MyClass { @Autowired private WebClient.Builder webClientBuilder; public Mono<String> doOtherStuff() { return webClientBuilder.build().get().uri("http://stores/stores") .retrieve().bodyToMono(String.class); } }
The URI needs to use a virtual host name (that is, a service name, not a host name). The Ribbon client is used to create a full physical address.
A load-balanced RestTemplate
can be configured to retry failed requests.
By default, this logic is disabled.
You can enable it by adding Spring Retry to your application’s classpath.
The load-balanced RestTemplate
honors some of the Ribbon configuration values related to retrying failed requests.
You can use client.ribbon.MaxAutoRetries
, client.ribbon.MaxAutoRetriesNextServer
, and client.ribbon.OkToRetryOnAllOperations
properties.
If you would like to disable the retry logic with Spring Retry on the classpath, you can set spring.cloud.loadbalancer.retry.enabled=false
.
See the Ribbon documentation for a description of what these properties do.
If you would like to implement a BackOffPolicy
in your retries, you need to create a bean of type LoadBalancedRetryFactory
and override the createBackOffPolicy
method:
@Configuration public class MyConfiguration { @Bean LoadBalancedRetryFactory retryFactory() { return new LoadBalancedRetryFactory() { @Override public BackOffPolicy createBackOffPolicy(String service) { return new ExponentialBackOffPolicy(); } }; } }
![]() | Note |
---|---|
|
If you want to add one or more RetryListener
implementations to your retry functionality, you need to
create a bean of type LoadBalancedRetryListenerFactory
and return the RetryListener
array
you would like to use for a given service, as shown in the following example:
@Configuration public class MyConfiguration { @Bean LoadBalancedRetryListenerFactory retryListenerFactory() { return new LoadBalancedRetryListenerFactory() { @Override public RetryListener[] createRetryListeners(String service) { return new RetryListener[]{new RetryListener() { @Override public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) { //TODO Do you business... return true; } @Override public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) { //TODO Do you business... } @Override public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) { //TODO Do you business... } }}; } }; } }
If you want a RestTemplate
that is not load-balanced, create a RestTemplate
bean and inject it.
To access the load-balanced RestTemplate
, use the @LoadBalanced
qualifier when you create your @Bean
, as shown in the following example:\
@Configuration public class MyConfiguration { @LoadBalanced @Bean RestTemplate loadBalanced() { return new RestTemplate(); } @Primary @Bean RestTemplate restTemplate() { return new RestTemplate(); } } public class MyClass { @Autowired private RestTemplate restTemplate; @Autowired @LoadBalanced private RestTemplate loadBalanced; public String doOtherStuff() { return loadBalanced.getForObject("http://stores/stores", String.class); } public String doStuff() { return restTemplate.getForObject("http://example.com", String.class); } }
![]() | Important |
---|---|
Notice the use of the |
![]() | Tip |
---|---|
If you see errors such as |
WebClient
can be configured to use the LoadBalancerClient
. LoadBalancerExchangeFilterFunction
is auto-configured if spring-webflux
is on the classpath. The following example shows how to configure a WebClient
to use load balancer:
public class MyClass { @Autowired private LoadBalancerExchangeFilterFunction lbFunction; public Mono<String> doOtherStuff() { return WebClient.builder().baseUrl("http://stores") .filter(lbFunction) .build() .get() .uri("/stores") .retrieve() .bodyToMono(String.class); } }
The URI needs to use a virtual host name (that is, a service name, not a host name).
The LoadBalancerClient
is used to create a full physical address.
Sometimes, it is useful to ignore certain named network interfaces so that they can be excluded from Service Discovery registration (for example, when running in a Docker container).
A list of regular expressions can be set to cause the desired network interfaces to be ignored.
The following configuration ignores the docker0
interface and all interfaces that start with veth
:
application.yml.
spring: cloud: inetutils: ignoredInterfaces: - docker0 - veth.*
You can also force the use of only specified network addresses by using a list of regular expressions, as shown in the following example:
bootstrap.yml.
spring: cloud: inetutils: preferredNetworks: - 192.168 - 10.0
You can also force the use of only site-local addresses, as shown in the following example: .application.yml
spring: cloud: inetutils: useOnlySiteLocalInterfaces: true
See Inet4Address.html.isSiteLocalAddress() for more details about what constitutes a site-local address.
Spring Cloud Commons provides beans for creating both Apache HTTP clients (ApacheHttpClientFactory
) and OK HTTP clients (OkHttpClientFactory
).
The OkHttpClientFactory
bean is created only if the OK HTTP jar is on the classpath.
In addition, Spring Cloud Commons provides beans for creating the connection managers used by both clients: ApacheHttpClientConnectionManagerFactory
for the Apache HTTP client and OkHttpClientConnectionPoolFactory
for the OK HTTP client.
If you would like to customize how the HTTP clients are created in downstream projects, you can provide your own implementation of these beans.
In addition, if you provide a bean of type HttpClientBuilder
or OkHttpClient.Builder
, the default factories use these builders as the basis for the builders returned to downstream projects.
You can also disable the creation of these beans by setting spring.cloud.httpclientfactories.apache.enabled
or spring.cloud.httpclientfactories.ok.enabled
to false
.
Spring Cloud Commons provides a /features
actuator endpoint.
This endpoint returns features available on the classpath and whether they are enabled.
The information returned includes the feature type, name, version, and vendor.
There are two types of 'features': abstract and named.
Abstract features are features where an interface or abstract class is defined and that an implementation the creates, such as DiscoveryClient
, LoadBalancerClient
, or LockService
.
The abstract class or interface is used to find a bean of that type in the context.
The version displayed is bean.getClass().getPackage().getImplementationVersion()
.
Named features are features that do not have a particular class they implement, such as "Circuit Breaker", "API Gateway", "Spring Cloud Bus", and others. These features require a name and a bean type.
Any module can declare any number of HasFeature
beans, as shown in the following examples:
@Bean public HasFeatures commonsFeatures() { return HasFeatures.abstractFeatures(DiscoveryClient.class, LoadBalancerClient.class); } @Bean public HasFeatures consulFeatures() { return HasFeatures.namedFeatures( new NamedFeature("Spring Cloud Bus", ConsulBusAutoConfiguration.class), new NamedFeature("Circuit Breaker", HystrixCommandAspect.class)); } @Bean HasFeatures localFeatures() { return HasFeatures.builder() .abstractFeature(Foo.class) .namedFeature(new NamedFeature("Bar Feature", Bar.class)) .abstractFeature(Baz.class) .build(); }
Each of these beans should go in an appropriately guarded @Configuration
.
Due to the fact that some users have problem with setting up Spring Cloud application, we’ve decided to add a compatibility verification mechanism. It will break if your current setup is not compatible with Spring Cloud requirements, together with a report, showing what exactly went wrong.
At the moment we verify which version of Spring Boot is added to your classpath.
Example of a report
*************************** APPLICATION FAILED TO START *************************** Description: Your project setup is incompatible with our requirements due to following reasons: - Spring Boot [2.1.0.RELEASE] is not compatible with this Spring Cloud release train Action: Consider applying the following actions: - Change Spring Boot version to one of the following versions [1.2.x, 1.3.x] . You can find the latest Spring Boot versions here [https://spring.io/projects/spring-boot#learn]. If you want to learn more about the Spring Cloud Release train compatibility, you can visit this page [https://spring.io/projects/spring-cloud#overview] and check the [Release Trains] section.
In order to disable this feature, set spring.cloud.compatibility-verifier.enabled
to false
.
If you want to override the compatible Spring Boot versions, just set the
spring.cloud.compatibility-verifier.compatible-boot-versions
property with a comma separated list
of compatible Spring Boot versions.
1.0.0.BUILD-SNAPSHOT
Spring Cloud Config provides server-side and client-side support for externalized configuration in a distributed system. With the Config Server, you have a central place to manage external properties for applications across all environments.
The concepts on both client and server map identically to the Spring Environment
and PropertySource
abstractions, so they fit very well with Spring applications but can be used with any application running in any language.
As an application moves through the deployment pipeline from dev to test and into production, you can manage the configuration between those environments and be certain that applications have everything they need to run when they migrate.
The default implementation of the server storage backend uses git, so it easily supports labelled versions of configuration environments as well as being accessible to a wide range of tooling for managing the content.
It is easy to add alternative implementations and plug them in with Spring configuration.
This quick start walks through using both the server and the client of Spring Cloud Config Server.
First, start the server, as follows:
$ cd spring-cloud-config-server $ ../mvnw spring-boot:run
The server is a Spring Boot application, so you can run it from your IDE if you prefer to do so (the main class is ConfigServerApplication
).
Next try out a client, as follows:
$ curl localhost:8888/foo/development {"name":"foo","label":"master","propertySources":[ {"name":"https://github.com/scratches/config-repo/foo-development.properties","source":{"bar":"spam"}}, {"name":"https://github.com/scratches/config-repo/foo.properties","source":{"foo":"bar"}} ]}
The default strategy for locating property sources is to clone a git repository (at spring.cloud.config.server.git.uri
) and use it to initialize a mini SpringApplication
.
The mini-application’s Environment
is used to enumerate property sources and publish them at a JSON endpoint.
The HTTP service has resources in the following form:
/{application}/{profile}[/{label}] /{application}-{profile}.yml /{label}/{application}-{profile}.yml /{application}-{profile}.properties /{label}/{application}-{profile}.properties
where application
is injected as the spring.config.name
in the SpringApplication
(what is normally application
in a regular Spring Boot app), profile
is an active profile (or comma-separated list of properties), and label
is an optional git label (defaults to master
.)
Spring Cloud Config Server pulls configuration for remote clients from various sources. The following example gets configuration from a git repository (which must be provided), as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo
Other sources are any JDBC compatible database, Subversion, Hashicorp Vault, Credhub and local filesystems.
To use these features in an application, you can build it as a Spring Boot application that depends on spring-cloud-config-client (for an example, see the test cases for the config-client or the sample application).
The most convenient way to add the dependency is with a Spring Boot starter org.springframework.cloud:spring-cloud-starter-config
.
There is also a parent pom and BOM (spring-cloud-starter-parent
) for Maven users and a Spring IO version management properties file for Gradle and Spring CLI users. The following example shows a typical Maven configuration:
pom.xml.
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>{spring-boot-docs-version}</version> <relativePath /> <!-- lookup parent from repository --> </parent> <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-dependencies</artifactId> <version>{spring-cloud-version}</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-config</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> <!-- repositories also needed for snapshots and milestones -->
Now you can create a standard Spring Boot application, such as the following HTTP server:
@SpringBootApplication @RestController public class Application { @RequestMapping("/") public String home() { return "Hello World!"; } public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
When this HTTP server runs, it picks up the external configuration from the default local config server (if it is running) on port 8888.
To modify the startup behavior, you can change the location of the config server by using bootstrap.properties
(similar to application.properties
but for the bootstrap phase of an application context), as shown in the following example:
spring.cloud.config.uri: http://myconfigserver.com
By default, if no application name is set, application
will be used. To modify the name, the following property can be added to the bootstrap.properties
file:
spring.application.name: myapp
![]() | Note |
---|---|
When setting the property |
The bootstrap properties show up in the /env
endpoint as a high-priority property source, as shown in the following example.
$ curl localhost:8080/env { "profiles":[], "configService:https://github.com/spring-cloud-samples/config-repo/bar.properties":{"foo":"bar"}, "servletContextInitParams":{}, "systemProperties":{...}, ... }
A property source called ``configService:<URL of remote repository>/<file name>
contains the foo
property with a value of bar
and is highest priority.
![]() | Note |
---|---|
The URL in the property source name is the git repository, not the config server URL. |
Spring Cloud Config Server provides an HTTP resource-based API for external configuration (name-value pairs or equivalent YAML content).
The server is embeddable in a Spring Boot application, by using the @EnableConfigServer
annotation.
Consequently, the following application is a config server:
ConfigServer.java.
@SpringBootApplication @EnableConfigServer public class ConfigServer { public static void main(String[] args) { SpringApplication.run(ConfigServer.class, args); } }
Like all Spring Boot applications, it runs on port 8080 by default, but you can switch it to the more conventional port 8888 in various ways.
The easiest, which also sets a default configuration repository, is by launching it with spring.config.name=configserver
(there is a configserver.yml
in the Config Server jar).
Another is to use your own application.properties
, as shown in the following example:
application.properties.
server.port: 8888 spring.cloud.config.server.git.uri: file://${user.home}/config-repo
where ${user.home}/config-repo
is a git repository containing YAML and properties files.
![]() | Note |
---|---|
On Windows, you need an extra "/" in the file URL if it is absolute with a drive prefix (for example, |
![]() | Tip |
---|---|
The following listing shows a recipe for creating the git repository in the preceding example: $ cd $HOME $ mkdir config-repo $ cd config-repo $ git init . $ echo info.foo: bar > application.properties $ git add -A . $ git commit -m "Add application.properties" |
![]() | Warning |
---|---|
Using the local filesystem for your git repository is intended for testing only. You should use a server to host your configuration repositories in production. |
![]() | Warning |
---|---|
The initial clone of your configuration repository can be quick and efficient if you keep only text files in it. If you store binary files, especially large ones, you may experience delays on the first request for configuration or encounter out of memory errors in the server. |
Where should you store the configuration data for the Config Server?
The strategy that governs this behaviour is the EnvironmentRepository
, serving Environment
objects.
This Environment
is a shallow copy of the domain from the Spring Environment
(including propertySources
as the main feature).
The Environment
resources are parametrized by three variables:
{application}
, which maps to spring.application.name
on the client side.{profile}
, which maps to spring.profiles.active
on the client (comma-separated list).{label}
, which is a server side feature labelling a "versioned" set of config files.Repository implementations generally behave like a Spring Boot application, loading configuration files from a spring.config.name
equal to the {application}
parameter, and spring.profiles.active
equal to the {profiles}
parameter.
Precedence rules for profiles are also the same as in a regular Spring Boot application: Active profiles take precedence over defaults, and, if there are multiple profiles, the last one wins (similar to adding entries to a Map
).
The following sample client application has this bootstrap configuration:
bootstrap.yml.
spring: application: name: foo profiles: active: dev,mysql
(As usual with a Spring Boot application, these properties could also be set by environment variables or command line arguments).
If the repository is file-based, the server creates an
Environment
from application.yml
(shared between all clients) and
foo.yml
(with foo.yml
taking precedence).
If the YAML files have documents inside them that point to Spring profiles, those are applied with higher precedence (in order of the profiles listed).
If there are profile-specific YAML (or properties) files, these are also applied with higher precedence than the defaults.
Higher precedence translates to a PropertySource
listed earlier in the Environment
.
(These same rules apply in a standalone Spring Boot application.)
You can set spring.cloud.config.server.accept-empty to false so that Server would return a HTTP 404 status, if the application is not found.By default, this flag is set to true.
The default implementation of EnvironmentRepository
uses a Git backend, which is very convenient for managing upgrades and physical environments and for auditing changes.
To change the location of the repository, you can set the spring.cloud.config.server.git.uri
configuration property in the Config Server (for example in application.yml
).
If you set it with a file:
prefix, it should work from a local repository so that you can get started quickly and easily without a server. However, in that case, the server operates directly on the local repository without cloning it (it does not matter if it is not bare because the Config Server never makes changes to the "remote" repository).
To scale the Config Server up and make it highly available, you need to have all instances of the server pointing to the same repository, so only a shared file system would work.
Even in that case, it is better to use the ssh:
protocol for a shared filesystem repository, so that the server can clone it and use a local working copy as a cache.
This repository implementation maps the {label}
parameter of the HTTP resource to a git label (commit id, branch name, or tag).
If the git branch or tag name contains a slash (/
), then the label in the HTTP URL should instead be specified with the special string (_)
(to avoid ambiguity with other URL paths).
For example, if the label is foo/bar
, replacing the slash would result in the following label: foo(_)bar
.
The inclusion of the special string (_)
can also be applied to the {application}
parameter.
If you use a command-line client such as curl, be careful with the brackets in the URL — you should escape them from the shell with single quotes ('').
The configuration server’s validation of the Git server’s SSL certificate can be disabled by setting the git.skipSslValidation
property to true
(default is false
).
spring: cloud: config: server: git: uri: https://example.com/my/repo skipSslValidation: true
You can configure the time, in seconds, that the configuration server will wait to acquire an HTTP connection. Use the git.timeout
property.
spring: cloud: config: server: git: uri: https://example.com/my/repo timeout: 4
Spring Cloud Config Server supports a git repository URL with placeholders for the {application}
and {profile}
(and {label}
if you need it, but remember that the label is applied as a git label anyway).
So you can support a “one repository per application” policy by using a structure similar to the following:
spring: cloud: config: server: git: uri: https://github.com/myorg/{application}
You can also support a “one repository per profile” policy by using a similar pattern but with
{profile}
.
Additionally, using the special string "(_)" within your {application}
parameters can enable support for multiple
organizations, as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/{application}
where {application}
is provided at request time in the following format: organization(_)application
.
Spring Cloud Config also includes support for more complex requirements with pattern
matching on the application and profile name.
The pattern format is a comma-separated list of {application}/{profile}
names with wildcards (note that a pattern beginning with a wildcard may need to be quoted), as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo repos: simple: https://github.com/simple/config-repo special: pattern: special*/dev*,*special*/dev* uri: https://github.com/special/config-repo local: pattern: local* uri: file:/home/configsvc/config-repo
If {application}/{profile}
does not match any of the patterns, it uses the default URI defined under spring.cloud.config.server.git.uri
.
In the above example, for the “simple” repository, the pattern is simple/*
(it only matches one application named simple
in all profiles). The “local” repository matches all application names beginning with local
in all profiles (the /*
suffix is added automatically to any pattern that does not have a profile matcher).
![]() | Note |
---|---|
The “one-liner” short cut used in the “simple” example can be used only if the only property to be set is the URI. If you need to set anything else (credentials, pattern, and so on) you need to use the full form. |
The pattern
property in the repo is actually an array, so you can use a YAML array (or [0]
, [1]
, etc. suffixes in properties files) to bind to multiple patterns.
You may need to do so if you are going to run apps with multiple profiles, as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo repos: development: pattern: - '*/development' - '*/staging' uri: https://github.com/development/config-repo staging: pattern: - '*/qa' - '*/production' uri: https://github.com/staging/config-repo
![]() | Note |
---|---|
Spring Cloud guesses that a pattern containing a profile that does not end in |
Every repository can also optionally store config files in sub-directories, and patterns to search for those directories can be specified as searchPaths
.
The following example shows a config file at the top level:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo searchPaths: foo,bar*
In the preceding example, the server searches for config files in the top level and in the foo/
sub-directory and also any sub-directory whose name begins with bar
.
By default, the server clones remote repositories when configuration is first requested. The server can be configured to clone the repositories at startup, as shown in the following top-level example:
spring: cloud: config: server: git: uri: https://git/common/config-repo.git repos: team-a: pattern: team-a-* cloneOnStart: true uri: http://git/team-a/config-repo.git team-b: pattern: team-b-* cloneOnStart: false uri: http://git/team-b/config-repo.git team-c: pattern: team-c-* uri: http://git/team-a/config-repo.git
In the preceding example, the server clones team-a’s config-repo on startup, before it accepts any requests. All other repositories are not cloned until configuration from the repository is requested.
![]() | Note |
---|---|
Setting a repository to be cloned when the Config Server starts up can help to identify a misconfigured configuration source (such as an invalid repository URI) quickly, while the Config Server is starting up.
With |
To use HTTP basic authentication on the remote repository, add the username
and password
properties separately (not in the URL), as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo username: trolley password: strongpassword
If you do not use HTTPS and user credentials, SSH should also work out of the box when you store keys in the default directories (~/.ssh
) and the URI points to an SSH location, such as [email protected]:configuration/cloud-configuration
.
It is important that an entry for the Git server be present in the ~/.ssh/known_hosts
file and that it is in ssh-rsa
format.
Other formats (such as ecdsa-sha2-nistp256
) are not supported.
To avoid surprises, you should ensure that only one entry is present in the known_hosts
file for the Git server and that it matches the URL you provided to the config server.
If you use a hostname in the URL, you want to have exactly that (not the IP) in the known_hosts
file.
The repository is accessed by using JGit, so any documentation you find on that should be applicable.
HTTPS proxy settings can be set in ~/.git/config
or (in the same way as for any other JVM process) with
system properties (-Dhttps.proxyHost
and -Dhttps.proxyPort
).
![]() | Tip |
---|---|
If you do not know where your |
Spring Cloud Config Server also supports AWS CodeCommit authentication. AWS CodeCommit uses an authentication helper when using Git from the command line. This helper is not used with the JGit library, so a JGit CredentialProvider for AWS CodeCommit is created if the Git URI matches the AWS CodeCommit pattern. AWS CodeCommit URIs follow this pattern://git-codecommit.${AWS_REGION}.amazonaws.com/${repopath}.
If you provide a username and password with an AWS CodeCommit URI, they must be the AWS accessKeyId and secretAccessKey that provide access to the repository. If you do not specify a username and password, the accessKeyId and secretAccessKey are retrieved by using the AWS Default Credential Provider Chain.
If your Git URI matches the CodeCommit URI pattern (shown earlier), you must provide valid AWS credentials in the username and password or in one of the locations supported by the default credential provider chain. AWS EC2 instances may use IAM Roles for EC2 Instances.
![]() | Note |
---|---|
The |
By default, the JGit library used by Spring Cloud Config Server uses SSH configuration files such as ~/.ssh/known_hosts
and /etc/ssh/ssh_config
when connecting to Git repositories by using an SSH URI.
In cloud environments such as Cloud Foundry, the local filesystem may be ephemeral or not easily accessible.
For those cases, SSH configuration can be set by using Java properties.
In order to activate property-based SSH configuration, the spring.cloud.config.server.git.ignoreLocalSshSettings
property must be set to true
, as shown in the following example:
spring: cloud: config: server: git: uri: git@gitserver.com:team/repo1.git ignoreLocalSshSettings: true hostKey: someHostKey hostKeyAlgorithm: ssh-rsa privateKey: | -----BEGIN RSA PRIVATE KEY----- MIIEpgIBAAKCAQEAx4UbaDzY5xjW6hc9jwN0mX33XpTDVW9WqHp5AKaRbtAC3DqX IXFMPgw3K45jxRb93f8tv9vL3rD9CUG1Gv4FM+o7ds7FRES5RTjv2RT/JVNJCoqF ol8+ngLqRZCyBtQN7zYByWMRirPGoDUqdPYrj2yq+ObBBNhg5N+hOwKjjpzdj2Ud 1l7R+wxIqmJo1IYyy16xS8WsjyQuyC0lL456qkd5BDZ0Ag8j2X9H9D5220Ln7s9i oezTipXipS7p7Jekf3Ywx6abJwOmB0rX79dV4qiNcGgzATnG1PkXxqt76VhcGa0W DDVHEEYGbSQ6hIGSh0I7BQun0aLRZojfE3gqHQIDAQABAoIBAQCZmGrk8BK6tXCd fY6yTiKxFzwb38IQP0ojIUWNrq0+9Xt+NsypviLHkXfXXCKKU4zUHeIGVRq5MN9b BO56/RrcQHHOoJdUWuOV2qMqJvPUtC0CpGkD+valhfD75MxoXU7s3FK7yjxy3rsG EmfA6tHV8/4a5umo5TqSd2YTm5B19AhRqiuUVI1wTB41DjULUGiMYrnYrhzQlVvj 5MjnKTlYu3V8PoYDfv1GmxPPh6vlpafXEeEYN8VB97e5x3DGHjZ5UrurAmTLTdO8 +AahyoKsIY612TkkQthJlt7FJAwnCGMgY6podzzvzICLFmmTXYiZ/28I4BX/mOSe pZVnfRixAoGBAO6Uiwt40/PKs53mCEWngslSCsh9oGAaLTf/XdvMns5VmuyyAyKG ti8Ol5wqBMi4GIUzjbgUvSUt+IowIrG3f5tN85wpjQ1UGVcpTnl5Qo9xaS1PFScQ xrtWZ9eNj2TsIAMp/svJsyGG3OibxfnuAIpSXNQiJPwRlW3irzpGgVx/AoGBANYW dnhshUcEHMJi3aXwR12OTDnaLoanVGLwLnkqLSYUZA7ZegpKq90UAuBdcEfgdpyi PhKpeaeIiAaNnFo8m9aoTKr+7I6/uMTlwrVnfrsVTZv3orxjwQV20YIBCVRKD1uX VhE0ozPZxwwKSPAFocpyWpGHGreGF1AIYBE9UBtjAoGBAI8bfPgJpyFyMiGBjO6z FwlJc/xlFqDusrcHL7abW5qq0L4v3R+FrJw3ZYufzLTVcKfdj6GelwJJO+8wBm+R gTKYJItEhT48duLIfTDyIpHGVm9+I1MGhh5zKuCqIhxIYr9jHloBB7kRm0rPvYY4 VAykcNgyDvtAVODP+4m6JvhjAoGBALbtTqErKN47V0+JJpapLnF0KxGrqeGIjIRV cYA6V4WYGr7NeIfesecfOC356PyhgPfpcVyEztwlvwTKb3RzIT1TZN8fH4YBr6Ee KTbTjefRFhVUjQqnucAvfGi29f+9oE3Ei9f7wA+H35ocF6JvTYUsHNMIO/3gZ38N CPjyCMa9AoGBAMhsITNe3QcbsXAbdUR00dDsIFVROzyFJ2m40i4KCRM35bC/BIBs q0TY3we+ERB40U8Z2BvU61QuwaunJ2+uGadHo58VSVdggqAo0BSkH58innKKt96J 69pcVH/4rmLbXdcmNYGm6iu+MlPQk4BUZknHSmVHIFdJ0EPupVaQ8RHT -----END RSA PRIVATE KEY-----
The following table describes the SSH configuration properties.
Table 5.1. SSH Configuration Properties
Property Name | Remarks |
---|---|
ignoreLocalSshSettings | If |
privateKey | Valid SSH private key. Must be set if |
hostKey | Valid SSH host key. Must be set if |
hostKeyAlgorithm | One of |
strictHostKeyChecking |
|
knownHostsFile | Location of custom |
preferredAuthentications | Override server authentication method order. This should allow for evading login prompts if server has keyboard-interactive authentication before the |
Spring Cloud Config Server also supports a search path with placeholders for the {application}
and {profile}
(and {label}
if
you need it), as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo searchPaths: '{application}'
The preceding listing causes a search of the repository for files in the same name as the directory (as well as the top level). Wildcards are also valid in a search path with placeholders (any matching directory is included in the search).
As mentioned earlier, Spring Cloud Config Server makes a clone of the remote git repository in case the local copy gets dirty (for example, folder content changes by an OS process) such that Spring Cloud Config Server cannot update the local copy from remote repository.
To solve this issue, there is a force-pull
property that makes Spring Cloud Config Server force pull from the remote repository if the local copy is dirty, as shown in the following example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo force-pull: true
If you have a multiple-repositories configuration, you can configure the force-pull
property per repository, as shown in the following example:
spring: cloud: config: server: git: uri: https://git/common/config-repo.git force-pull: true repos: team-a: pattern: team-a-* uri: http://git/team-a/config-repo.git force-pull: true team-b: pattern: team-b-* uri: http://git/team-b/config-repo.git force-pull: true team-c: pattern: team-c-* uri: http://git/team-a/config-repo.git
![]() | Note |
---|---|
The default value for |
As Spring Cloud Config Server has a clone of the remote git repository
after check-outing branch to local repo (e.g fetching properties by label) it will keep this branch
forever or till the next server restart (which creates new local repo).
So there could be a case when remote branch is deleted but local copy of it is still available for fetching.
And if Spring Cloud Config Server client service starts with --spring.cloud.config.label=deletedRemoteBranch,master
it will fetch properties from deletedRemoteBranch
local branch, but not from master
.
In order to keep local repository branches clean and up to remote - deleteUntrackedBranches
property could be set.
It will make Spring Cloud Config Server force delete untracked branches from local repository.
Example:
spring: cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo deleteUntrackedBranches: true
![]() | Note |
---|---|
The default value for |
You can control how often the config server will fetch updated configuration data
from your Git backend by using spring.cloud.config.server.git.refreshRate
. The
value of this property is specified in seconds. By default the value is 0, meaning
the config server will fetch updated configuration from the Git repo every time it
is requested.
![]() | Warning |
---|---|
With VCS-based backends (git, svn), files are checked out or cloned to the local filesystem.
By default, they are put in the system temporary directory with a prefix of |
There is also a “native” profile in the Config Server that does not use Git but loads the config files from the local classpath or file system (any static URL you want to point to with spring.cloud.config.server.native.searchLocations
).
To use the native profile, launch the Config Server with spring.profiles.active=native
.
![]() | Note |
---|---|
Remember to use the |
![]() | Warning |
---|---|
The default value of the |
![]() | Tip |
---|---|
A filesystem backend is great for getting started quickly and for testing. To use it in production, you need to be sure that the file system is reliable and shared across all instances of the Config Server. |
The search locations can contain placeholders for {application}
, {profile}
, and {label}
.
In this way, you can segregate the directories in the path and choose a strategy that makes sense for you (such as subdirectory per application or subdirectory per profile).
If you do not use placeholders in the search locations, this repository also appends the {label}
parameter of the HTTP resource to a suffix on the search path, so properties files are loaded from each search location and a subdirectory with the same name as the label (the labelled properties take precedence in the Spring Environment).
Thus, the default behaviour with no placeholders is the same as adding a search location ending with /{label}/
.
For example, file:/tmp/config
is the same as file:/tmp/config,file:/tmp/config/{label}
.
This behavior can be disabled by setting spring.cloud.config.server.native.addLabelLocations=false
.
Spring Cloud Config Server also supports Vault as a backend.
For more information on Vault, see the Vault quick start guide.
To enable the config server to use a Vault backend, you can run your config server with the vault
profile.
For example, in your config server’s application.properties
, you can add spring.profiles.active=vault
.
By default, the config server assumes that your Vault server runs at http://127.0.0.1:8200
.
It also assumes that the name of backend is secret
and the key is application
.
All of these defaults can be configured in your config server’s application.properties
.
The following table describes configurable Vault properties:
Name | Default Value |
---|---|
host | 127.0.0.1 |
port | 8200 |
scheme | http |
backend | secret |
defaultKey | application |
profileSeparator | , |
kvVersion | 1 |
skipSslValidation | false |
timeout | 5 |
namespace | null |
![]() | Important |
---|---|
All of the properties in the preceding table must be prefixed with |
All configurable properties can be found in org.springframework.cloud.config.server.environment.VaultEnvironmentProperties
.
Vault 0.10.0 introduced a versioned key-value backend (k/v backend version 2) that exposes a different API than earlier versions, it now requires a data/
between the mount path and the actual context path and wraps secrets in a data
object. Setting kvVersion=2
will take this into account.
Optionally, there is support for the Vault Enterprise X-Vault-Namespace
header. To have it sent to Vault set the namespace
property.
With your config server running, you can make HTTP requests to the server to retrieve values from the Vault backend. To do so, you need a token for your Vault server.
First, place some data in you Vault, as shown in the following example:
$ vault kv put secret/application foo=bar baz=bam $ vault kv put secret/myapp foo=myappsbar
Second, make an HTTP request to your config server to retrieve the values, as shown in the following example:
$ curl -X "GET" "http://localhost:8888/myapp/default" -H "X-Config-Token: yourtoken"
You should see a response similar to the following:
{ "name":"myapp", "profiles":[ "default" ], "label":null, "version":null, "state":null, "propertySources":[ { "name":"vault:myapp", "source":{ "foo":"myappsbar" } }, { "name":"vault:application", "source":{ "baz":"bam", "foo":"bar" } } ] }
When using Vault, you can provide your applications with multiple properties sources. For example, assume you have written data to the following paths in Vault:
secret/myApp,dev secret/myApp secret/application,dev secret/application
Properties written to secret/application
are available to all applications using the Config Server.
An application with the name, myApp
, would have any properties written to secret/myApp
and secret/application
available to it.
When myApp
has the dev
profile enabled, properties written to all of the above paths would be available to it, with properties in the first path in the list taking priority over the others.
The configuration server can access a Git or Vault backend through an HTTP or HTTPS proxy. This behavior is controlled for either Git or Vault by settings under proxy.http
and proxy.https
. These settings are per repository, so if you are using a composite environment repository you must configure proxy settings for each backend in the composite individually. If using a network which requires separate proxy servers for HTTP and HTTPS URLs, you can configure both the HTTP and the HTTPS proxy settings for a single backend.
The following table describes the proxy configuration properties for both HTTP and HTTPS proxies. All of these properties must be prefixed by proxy.http
or proxy.https
.
Table 5.2. Proxy Configuration Properties
Property Name | Remarks |
---|---|
host | The host of the proxy. |
port | The port with which to access the proxy. |
nonProxyHosts | Any hosts which the configuration server should access outside the proxy. If values are provided for both |
username | The username with which to authenticate to the proxy. If values are provided for both |
password | The password with which to authenticate to the proxy. If values are provided for both |
The following configuration uses an HTTPS proxy to access a Git repository.
spring: profiles: active: git cloud: config: server: git: uri: https://github.com/spring-cloud-samples/config-repo proxy: https: host: my-proxy.host.io password: myproxypassword port: '3128' username: myproxyusername nonProxyHosts: example.com
Sharing configuration between all applications varies according to which approach you take, as described in the following topics:
With file-based (git, svn, and native) repositories, resources with file names in application*
(application.properties
, application.yml
, application-*.properties
, and so on) are shared between all client applications.
You can use resources with these file names to configure global defaults and have them be overridden by application-specific files as necessary.
The #_property_overrides[property overrides] feature can also be used for setting global defaults, with placeholders applications allowed to override them locally.
![]() | Tip |
---|---|
With the “native” profile (a local file system backend) , you should use an explicit search location that is not part of the server’s own configuration.
Otherwise, the |
When using Vault as a backend, you can share configuration with all applications by placing configuration in secret/application
.
For example, if you run the following Vault command, all applications using the config server will have the properties foo
and baz
available to them:
$ vault write secret/application foo=bar baz=bam
Spring Cloud Config Server supports JDBC (relational database) as a backend for configuration properties.
You can enable this feature by adding spring-jdbc
to the classpath and using the jdbc
profile or by adding a bean of type JdbcEnvironmentRepository
.
If you include the right dependencies on the classpath (see the user guide for more details on that), Spring Boot configures a data source.
The database needs to have a table called PROPERTIES
with columns called APPLICATION
, PROFILE
, and LABEL
(with the usual Environment
meaning), plus KEY
and VALUE
for the key and value pairs in Properties
style.
All fields are of type String in Java, so you can make them VARCHAR
of whatever length you need.
Property values behave in the same way as they would if they came from Spring Boot properties files named {application}-{profile}.properties
, including all the encryption and decryption, which will be applied as post-processing steps (that is, not in the repository implementation directly).
Spring Cloud Config Server supports CredHub as a backend for configuration properties. You can enable this feature by adding a dependency to Spring CredHub.
pom.xml.
<dependencies> <dependency> <groupId>org.springframework.credhub</groupId> <artifactId>spring-credhub-starter</artifactId> </dependency> </dependencies>
The following configuration uses mutual TLS to access a CredHub:
spring: profiles: active: credhub cloud: config: server: credhub: url: https://credhub:8844
The properties should be stored as JSON, such as:
credhub set --name "/demo-app/default/master/toggles" --type=json value: {"toggle.button": "blue", "toggle.link": "red"}
credhub set --name "/demo-app/default/master/abs" --type=json value: {"marketing.enabled": true, "external.enabled": false}
All client applications with the name spring.cloud.config.name=demo-app
will have the following properties available to them:
{ toggle.button: "blue", toggle.link: "red", marketing.enabled: true, external.enabled: false }
![]() | Note |
---|---|
When no profile is specified |
You can authenticate with OAuth 2.0 using UAA as a provider.
pom.xml.
<dependencies> <dependency> <groupId>org.springframework.security</groupId> <artifactId>spring-security-config</artifactId> </dependency> <dependency> <groupId>org.springframework.security</groupId> <artifactId>spring-security-oauth2-client</artifactId> </dependency> </dependencies>
The following configuration uses OAuth 2.0 and UAA to access a CredHub:
spring: profiles: active: credhub cloud: config: server: credhub: url: https://credhub:8844 oauth2: registration-id: credhub-client security: oauth2: client: registration: credhub-client: provider: uaa client-id: credhub_config_server client-secret: asecret authorization-grant-type: client_credentials provider: uaa: token-uri: https://uaa:8443/oauth/token
![]() | Note |
---|---|
The used UAA client-id should have |
In some scenarios, you may wish to pull configuration data from multiple environment repositories.
To do so, you can enable the composite
profile in your configuration server’s application properties or YAML file.
If, for example, you want to pull configuration data from a Subversion repository as well as two Git repositories, you can set the following properties for your configuration server:
spring: profiles: active: composite cloud: config: server: composite: - type: svn uri: file:///path/to/svn/repo - type: git uri: file:///path/to/rex/git/repo - type: git uri: file:///path/to/walter/git/repo
Using this configuration, precedence is determined by the order in which repositories are listed under the composite
key.
In the above example, the Subversion repository is listed first, so a value found in the Subversion repository will override values found for the same property in one of the Git repositories.
A value found in the rex
Git repository will be used before a value found for the same property in the walter
Git repository.
If you want to pull configuration data only from repositories that are each of distinct types, you can enable the corresponding profiles, rather than the composite
profile, in your configuration server’s application properties or YAML file.
If, for example, you want to pull configuration data from a single Git repository and a single HashiCorp Vault server, you can set the following properties for your configuration server:
spring: profiles: active: git, vault cloud: config: server: git: uri: file:///path/to/git/repo order: 2 vault: host: 127.0.0.1 port: 8200 order: 1
Using this configuration, precedence can be determined by an order
property.
You can use the order
property to specify the priority order for all your repositories.
The lower the numerical value of the order
property, the higher priority it has.
The priority order of a repository helps resolve any potential conflicts between repositories that contain values for the same properties.
![]() | Note |
---|---|
If your composite environment includes a Vault server as in the previous example, you must include a Vault token in every request made to the configuration server. See Vault Backend. |
![]() | Note |
---|---|
Any type of failure when retrieving values from an environment repository results in a failure for the entire composite environment. |
![]() | Note |
---|---|
When using a composite environment, it is important that all repositories contain the same labels.
If you have an environment similar to those in the preceding examples and you request configuration data with the |
In addition to using one of the environment repositories from Spring Cloud, you can also provide your own EnvironmentRepository
bean to be included as part of a composite environment.
To do so, your bean must implement the EnvironmentRepository
interface.
If you want to control the priority of your custom EnvironmentRepository
within the composite environment, you should also implement the Ordered
interface and override the getOrdered
method.
If you do not implement the Ordered
interface, your EnvironmentRepository
is given the lowest priority.
The Config Server has an “overrides” feature that lets the operator provide configuration properties to all applications.
The overridden properties cannot be accidentally changed by the application with the normal Spring Boot hooks.
To declare overrides, add a map of name-value pairs to spring.cloud.config.server.overrides
, as shown in the following example:
spring: cloud: config: server: overrides: foo: bar
The preceding examples causes all applications that are config clients to read foo=bar
, independent of their own configuration.
![]() | Note |
---|---|
A configuration system cannot force an application to use configuration data in any particular way. Consequently, overrides are not enforceable. However, they do provide useful default behavior for Spring Cloud Config clients. |
![]() | Tip |
---|---|
Normally, Spring environment placeholders with |
![]() | Note |
---|---|
In YAML, you do not need to escape the backslash itself. However, in properties files, you do need to escape the backslash, when you configure the overrides on the server. |
You can change the priority of all overrides in the client to be more like default values, letting applications supply their own values in environment variables or System properties, by setting the spring.cloud.config.overrideNone=true
flag (the default is false) in the remote repository.
Config Server comes with a Health Indicator that checks whether the configured EnvironmentRepository
is working.
By default, it asks the EnvironmentRepository
for an application named app
, the default
profile, and the default label provided by the EnvironmentRepository
implementation.
You can configure the Health Indicator to check more applications along with custom profiles and custom labels, as shown in the following example:
spring: cloud: config: server: health: repositories: myservice: label: mylabel myservice-dev: name: myservice profiles: development
You can disable the Health Indicator by setting spring.cloud.config.server.health.enabled=false
.
You can secure your Config Server in any way that makes sense to you (from physical network security to OAuth2 bearer tokens), because Spring Security and Spring Boot offer support for many security arrangements.
To use the default Spring Boot-configured HTTP Basic security, include Spring Security on the classpath (for example, through spring-boot-starter-security
).
The default is a username of user
and a randomly generated password. A random password is not useful in practice, so we recommend you configure the password (by setting spring.security.user.password
) and encrypt it (see below for instructions on how to do that).
![]() | Important |
---|---|
To use the encryption and decryption features you need the full-strength JCE installed in your JVM (it is not included by default). You can download the “Java Cryptography Extension (JCE) Unlimited Strength Jurisdiction Policy Files” from Oracle and follow the installation instructions (essentially, you need to replace the two policy files in the JRE lib/security directory with the ones that you downloaded). |
If the remote property sources contain encrypted content (values starting with {cipher}
), they are decrypted before sending to clients over HTTP.
The main advantage of this setup is that the property values need not be in plain text when they are “at rest” (for example, in a git repository).
If a value cannot be decrypted, it is removed from the property source and an additional property is added with the same key but prefixed with invalid
and a value that means “not applicable” (usually <n/a>
).
This is largely to prevent cipher text being used as a password and accidentally leaking.
If you set up a remote config repository for config client applications, it might contain an application.yml
similar to the following:
application.yml.
spring: datasource: username: dbuser password: '{cipher}FKSAJDFGYOS8F7GLHAKERGFHLSAJ'
Encrypted values in a .properties file must not be wrapped in quotes. Otherwise, the value is not decrypted. The following example shows values that would work:
application.properties.
spring.datasource.username: dbuser spring.datasource.password: {cipher}FKSAJDFGYOS8F7GLHAKERGFHLSAJ
You can safely push this plain text to a shared git repository, and the secret password remains protected.
The server also exposes /encrypt
and /decrypt
endpoints (on the assumption that these are secured and only accessed by authorized agents).
If you edit a remote config file, you can use the Config Server to encrypt values by POSTing to the /encrypt
endpoint, as shown in the following example:
$ curl localhost:8888/encrypt -d mysecret 682bc583f4641835fa2db009355293665d2647dade3375c0ee201de2a49f7bda
![]() | Note |
---|---|
If the value you encrypt has characters in it that need to be URL encoded, you should use the |
![]() | Tip |
---|---|
Be sure not to include any of the curl command statistics in the encrypted value. Outputting the value to a file can help avoid this problem. |
The inverse operation is also available through /decrypt
(provided the server is
configured with a symmetric key or a full key pair), as shown in the following example:
$ curl localhost:8888/decrypt -d 682bc583f4641835fa2db009355293665d2647dade3375c0ee201de2a49f7bda mysecret
![]() | Tip |
---|---|
If you testing with curl, then use |
Take the encrypted value and add the {cipher}
prefix before you put it in the YAML or properties file and before you commit and push it to a remote (potentially insecure) store.
The /encrypt
and /decrypt
endpoints also both accept paths in the form of /*/{name}/{profiles}
, which can be used to control cryptography on a per-application (name) and per-profile basis when clients call into the main environment resource.
![]() | Note |
---|---|
To control the cryptography in this granular way, you must also provide a |
The spring
command line client (with Spring Cloud CLI extensions
installed) can also be used to encrypt and decrypt, as shown in the following example:
$ spring encrypt mysecret --key foo 682bc583f4641835fa2db009355293665d2647dade3375c0ee201de2a49f7bda $ spring decrypt --key foo 682bc583f4641835fa2db009355293665d2647dade3375c0ee201de2a49f7bda mysecret
To use a key in a file (such as an RSA public key for encryption), prepend the key value with "@" and provide the file path, as shown in the following example:
$ spring encrypt mysecret --key @${HOME}/.ssh/id_rsa.pub AQAjPgt3eFZQXwt8tsHAVv/QHiY5sI2dRcR+...
![]() | Note |
---|---|
The |
The Config Server can use a symmetric (shared) key or an asymmetric one (RSA key pair).
The asymmetric choice is superior in terms of security, but it is often more convenient to use a symmetric key since it is a single property value to configure in the bootstrap.properties
.
To configure a symmetric key, you need to set encrypt.key
to a secret String (or use the ENCRYPT_KEY
environment variable to keep it out of plain-text configuration files).
![]() | Note |
---|---|
You cannot configure an asymmetric key using |
To configure an asymmetric key use a keystore (e.g. as
created by the keytool
utility that comes with the JDK). The
keystore properties are encrypt.keyStore.*
with *
equal to
Property | Description |
---|---|
| Contains a |
| Holds the password that unlocks the keystore |
| Identifies which key in the store to use |
| The type of KeyStore to create. Defaults to |
The encryption is done with the public key, and a private key is needed for decryption. Thus, in principle, you can configure only the public key in the server if you want to only encrypt (and are prepared to decrypt the values yourself locally with the private key). In practice, you might not want to do decrypt locally, because it spreads the key management process around all the clients, instead of concentrating it in the server. On the other hand, it can be a useful option if your config server is relatively insecure and only a handful of clients need the encrypted properties.
To create a keystore for testing, you can use a command resembling the following:
$ keytool -genkeypair -alias mytestkey -keyalg RSA \ -dname "CN=Web Server,OU=Unit,O=Organization,L=City,S=State,C=US" \ -keypass changeme -keystore server.jks -storepass letmein
Put the server.jks
file in the classpath (for instance) and then, in
your bootstrap.yml
, for the Config Server, create the following settings:
encrypt: keyStore: location: classpath:/server.jks password: letmein alias: mytestkey secret: changeme
In addition to the {cipher}
prefix in encrypted property values, the Config Server looks for zero or more {name:value}
prefixes before the start of the (Base64 encoded) cipher text.
The keys are passed to a TextEncryptorLocator
, which can do whatever logic it needs to locate a TextEncryptor
for the cipher.
If you have configured a keystore (encrypt.keystore.location
), the default locator looks for keys with aliases supplied by the key
prefix, with a cipher text like resembling the following:
foo: bar: `{cipher}{key:testkey}...`
The locator looks for a key named "testkey".
A secret can also be supplied by using a {secret:…}
value in the prefix.
However, if it is not supplied, the default is to use the keystore password (which is what you get when you build a keystore and do not specify a secret).
If you do supply a secret, you should also encrypt the secret using a custom SecretLocator
.
When the keys are being used only to encrypt a few bytes of configuration data (that is, they are not being used elsewhere), key rotation is hardly ever necessary on cryptographic grounds.
However, you might occasionally need to change the keys (for example, in the event of a security breach).
In that case, all the clients would need to change their source config files (for example, in git) and use a new {key:…}
prefix in all the ciphers.
Note that the clients need to first check that the key alias is available in the Config Server keystore.
![]() | Tip |
---|---|
If you want to let the Config Server handle all encryption as well as decryption, the |
Sometimes you want the clients to decrypt the configuration locally, instead of doing it in the server.
In that case, if you provide the encrypt.*
configuration to locate a key, you can still have /encrypt
and /decrypt
endpoints, but you need to explicitly switch off the decryption of outgoing properties by placing spring.cloud.config.server.encrypt.enabled=false
in bootstrap.[yml|properties]
.
If you do not care about the endpoints, it should work if you do not configure either the key or the enabled flag.
The default JSON format from the environment endpoints is perfect for consumption by Spring applications, because it maps directly onto the Environment
abstraction.
If you prefer, you can consume the same data as YAML or Java properties by adding a suffix (".yml", ".yaml" or ".properties") to the resource path.
This can be useful for consumption by applications that do not care about the structure of the JSON endpoints or the extra metadata they provide (for example, an application that is not using Spring might benefit from the simplicity of this approach).
The YAML and properties representations have an additional flag (provided as a boolean query parameter called resolvePlaceholders
) to signal that placeholders in the source documents (in the standard Spring ${…}
form) should be resolved in the output before rendering, where possible.
This is a useful feature for consumers that do not know about the Spring placeholder conventions.
![]() | Note |
---|---|
There are limitations in using the YAML or properties formats, mainly in relation to the loss of metadata. For example, the JSON is structured as an ordered list of property sources, with names that correlate with the source. The YAML and properties forms are coalesced into a single map, even if the origin of the values has multiple sources, and the names of the original source files are lost. Also, the YAML representation is not necessarily a faithful representation of the YAML source in a backing repository either. It is constructed from a list of flat property sources, and assumptions have to be made about the form of the keys. |
Instead of using the Environment
abstraction (or one of the alternative representations of it in YAML or properties format), your applications might need generic plain-text configuration files that are tailored to their environment.
The Config Server provides these through an additional endpoint at /{name}/{profile}/{label}/{path}
, where name
, profile
, and label
have the same meaning as the regular environment endpoint, but path
is a file name (such as log.xml
).
The source files for this endpoint are located in the same way as for the environment endpoints.
The same search path is used for properties and YAML files.
However, instead of aggregating all matching resources, only the first one to match is returned.
After a resource is located, placeholders in the normal format (${…}
) are resolved by using the effective Environment
for the supplied application name, profile, and label.
In this way, the resource endpoint is tightly integrated with the environment endpoints.
Consider the following example for a GIT or SVN repository:
application.yml nginx.conf
where nginx.conf
looks like this:
server { listen 80; server_name ${nginx.server.name}; }
and application.yml
like this:
nginx: server: name: example.com --- spring: profiles: development nginx: server: name: develop.com
The /foo/default/master/nginx.conf
resource might be as follows:
server { listen 80; server_name example.com; }
and /foo/development/master/nginx.conf
like this:
server { listen 80; server_name develop.com; }
![]() | Note |
---|---|
As with the source files for environment configuration, the |
![]() | Note |
---|---|
If you do not want to supply the |
The Config Server runs best as a standalone application.
However, if need be, you can embed it in another application.
To do so, use the @EnableConfigServer
annotation.
An optional property named spring.cloud.config.server.bootstrap
can be useful in this case.
It is a flag to indicate whether the server should configure itself from its own remote repository.
By default, the flag is off, because it can delay startup.
However, when embedded in another application, it makes sense to initialize the same way as any other application.
When setting spring.cloud.config.server.bootstrap
to true
you must also use a composite environment repository configuration.
For example
spring: application: name: configserver profiles: active: composite cloud: config: server: composite: - type: native search-locations: ${HOME}/Desktop/config bootstrap: true
![]() | Note |
---|---|
If you use the bootstrap flag, the config server needs to have its name and repository URI configured in |
To change the location of the server endpoints, you can (optionally) set spring.cloud.config.server.prefix
(for example, /config
), to serve the resources under a prefix.
The prefix should start but not end with a /
.
It is applied to the @RequestMappings
in the Config Server (that is, underneath the Spring Boot server.servletPath
and server.contextPath
prefixes).
If you want to read the configuration for an application directly from the backend repository (instead of from the config server), you
basically want an embedded config server with no endpoints.
You can switch off the endpoints entirely by not using the @EnableConfigServer
annotation (set spring.cloud.config.server.bootstrap=true
).
Many source code repository providers (such as Github, Gitlab, Gitea, Gitee, Gogs, or Bitbucket) notify you of changes in a repository through a webhook.
You can configure the webhook through the provider’s user interface as a URL and a set of events in which you are interested.
For instance, Github uses a POST to the webhook with a JSON body containing a list of commits and a header (X-Github-Event
) set to push
.
If you add a dependency on the spring-cloud-config-monitor
library and activate the Spring Cloud Bus in your Config Server, then a /monitor
endpoint is enabled.
When the webhook is activated, the Config Server sends a RefreshRemoteApplicationEvent
targeted at the applications it thinks might have changed.
The change detection can be strategized.
However, by default, it looks for changes in files that match the application name (for example, foo.properties
is targeted at the foo
application, while application.properties
is targeted at all applications).
The strategy to use when you want to override the behavior is PropertyPathNotificationExtractor
, which accepts the request headers and body as parameters and returns a list of file paths that changed.
The default configuration works out of the box with Github, Gitlab, Gitea, Gitee, Gogs or Bitbucket.
In addition to the JSON notifications from Github, Gitlab, Gitee, or Bitbucket, you can trigger a change notification by POSTing to /monitor
with form-encoded body parameters in the pattern of path={name}
.
Doing so broadcasts to applications matching the {name}
pattern (which can contain wildcards).
![]() | Note |
---|---|
The |
![]() | Note |
---|---|
The default configuration also detects filesystem changes in local git repositories. In that case, the webhook is not used. However, as soon as you edit a config file, a refresh is broadcast. |
A Spring Boot application can take immediate advantage of the Spring Config Server (or other external property sources provided by the application developer).
It also picks up some additional useful features related to Environment
change events.
The default behavior for any application that has the Spring Cloud Config Client on the classpath is as follows:
When a config client starts, it binds to the Config Server (through the spring.cloud.config.uri
bootstrap configuration property) and initializes Spring Environment
with remote property sources.
The net result of this behavior is that all client applications that want to consume the Config Server need a bootstrap.yml
(or an environment variable) with the server address set in spring.cloud.config.uri
(it defaults to "http://localhost:8888").
If you use a DiscoveryClient
implementation, such as Spring Cloud Netflix and Eureka Service Discovery or Spring Cloud Consul, you can have the Config Server register with the Discovery Service.
However, in the default “Config First” mode, clients cannot take advantage of the registration.
If you prefer to use DiscoveryClient
to locate the Config Server, you can do so by setting spring.cloud.config.discovery.enabled=true
(the default is false
).
The net result of doing so is that client applications all need a bootstrap.yml
(or an environment variable) with the appropriate discovery configuration.
For example, with Spring Cloud Netflix, you need to define the Eureka server address (for example, in eureka.client.serviceUrl.defaultZone
).
The price for using this option is an extra network round trip on startup, to locate the service registration.
The benefit is that, as long as the Discovery Service is a fixed point, the Config Server can change its coordinates.
The default service ID is configserver
, but you can change that on the client by setting spring.cloud.config.discovery.serviceId
(and on the server, in the usual way for a service, such as by setting spring.application.name
).
The discovery client implementations all support some kind of metadata map (for example, we have eureka.instance.metadataMap
for Eureka).
Some additional properties of the Config Server may need to be configured in its service registration metadata so that clients can connect correctly.
If the Config Server is secured with HTTP Basic, you can configure the credentials as user
and password
.
Also, if the Config Server has a context path, you can set configPath
.
For example, the following YAML file is for a Config Server that is a Eureka client:
bootstrap.yml.
eureka: instance: ... metadataMap: user: osufhalskjrtl password: lviuhlszvaorhvlo5847 configPath: /config
In some cases, you may want to fail startup of a service if it cannot connect to the Config Server.
If this is the desired behavior, set the bootstrap configuration property spring.cloud.config.fail-fast=true
to make the client halt with an Exception.
If you expect that the config server may occasionally be unavailable when your application starts, you can make it keep trying after a failure.
First, you need to set spring.cloud.config.fail-fast=true
.
Then you need to add spring-retry
and spring-boot-starter-aop
to your classpath.
The default behavior is to retry six times with an initial backoff interval of 1000ms and an exponential multiplier of 1.1 for subsequent backoffs.
You can configure these properties (and others) by setting the spring.cloud.config.retry.*
configuration properties.
![]() | Tip |
---|---|
To take full control of the retry behavior, add a |
The Config Service serves property sources from /{name}/{profile}/{label}
, where the default bindings in the client app are as follows:
${spring.application.name}
${spring.profiles.active}
(actually Environment.getActiveProfiles()
)![]() | Note |
---|---|
When setting the property |
You can override all of them by setting spring.cloud.config.*
(where *
is name
, profile
or label
).
The label
is useful for rolling back to previous versions of configuration.
With the default Config Server implementation, it can be a git label, branch name, or commit ID.
Label can also be provided as a comma-separated list.
In that case, the items in the list are tried one by one until one succeeds.
This behavior can be useful when working on a feature branch.
For instance, you might want to align the config label with your branch but make it optional (in that case, use spring.cloud.config.label=myfeature,develop
).
To ensure high availability when you have multiple instances of Config Server deployed and expect one or more instances to be unavailable from time to time, you can either specify multiple URLs (as a comma-separated list under the spring.cloud.config.uri
property) or have all your instances register in a Service Registry like Eureka ( if using Discovery-First Bootstrap mode ). Note that doing so ensures high availability only when the Config Server is not running (that is, when the application has exited) or when a connection timeout has occurred. For example, if the Config Server returns a 500 (Internal Server Error) response or the Config Client receives a 401 from the Config Server (due to bad credentials or other causes), the Config Client does not try to fetch properties from other URLs. An error of that kind indicates a user issue rather than an availability problem.
If you use HTTP basic security on your Config Server, it is currently possible to support per-Config Server auth credentials only if you embed the credentials in each URL you specify under the spring.cloud.config.uri
property. If you use any other kind of security mechanism, you cannot (currently) support per-Config Server authentication and authorization.
If you want to configure read timeout, this can be done by using the property spring.cloud.config.request-read-timeout
.
If you use HTTP Basic security on the server, clients need to know the password (and username if it is not the default). You can specify the username and password through the config server URI or via separate username and password properties, as shown in the following example:
bootstrap.yml.
spring: cloud: config: uri: https://user:[email protected]
The following example shows an alternate way to pass the same information:
bootstrap.yml.
spring: cloud: config: uri: https://myconfig.mycompany.com username: user password: secret
The spring.cloud.config.password
and spring.cloud.config.username
values override anything that is provided in the URI.
If you deploy your apps on Cloud Foundry, the best way to provide the password is through service credentials (such as in the URI, since it does not need to be in a config file).
The following example works locally and for a user-provided service on Cloud Foundry named configserver
:
bootstrap.yml.
spring: cloud: config: uri: ${vcap.services.configserver.credentials.uri:http://user:password@localhost:8888}
If you use another form of security, you might need to provide a RestTemplate
to the ConfigServicePropertySourceLocator
(for example, by grabbing it in the bootstrap context and injecting it).
The Config Client supplies a Spring Boot Health Indicator that attempts to load configuration from the Config Server.
The health indicator can be disabled by setting health.config.enabled=false
.
The response is also cached for performance reasons.
The default cache time to live is 5 minutes.
To change that value, set the health.config.time-to-live
property (in milliseconds).
In some cases, you might need to customize the requests made to the config server from the client.
Typically, doing so involves passing special Authorization
headers to authenticate requests to the server.
To provide a custom RestTemplate
:
PropertySourceLocator
, as shown in the following example:CustomConfigServiceBootstrapConfiguration.java.
@Configuration public class CustomConfigServiceBootstrapConfiguration { @Bean public ConfigServicePropertySourceLocator configServicePropertySourceLocator() { ConfigClientProperties clientProperties = configClientProperties(); ConfigServicePropertySourceLocator configServicePropertySourceLocator = new ConfigServicePropertySourceLocator(clientProperties); configServicePropertySourceLocator.setRestTemplate(customRestTemplate(clientProperties)); return configServicePropertySourceLocator; } }
resources/META-INF
, create a file called
spring.factories
and specify your custom configuration, as shown in the following example:spring.factories.
org.springframework.cloud.bootstrap.BootstrapConfiguration = com.my.config.client.CustomConfigServiceBootstrapConfiguration
When using Vault as a backend to your config server, the client needs to supply a token for the server to retrieve values from Vault.
This token can be provided within the client by setting spring.cloud.config.token
in bootstrap.yml
, as shown in the following example:
bootstrap.yml.
spring: cloud: config: token: YourVaultToken
Vault supports the ability to nest keys in a value stored in Vault, as shown in the following example:
echo -n '{"appA": {"secret": "appAsecret"}, "bar": "baz"}' | vault write secret/myapp -
This command writes a JSON object to your Vault.
To access these values in Spring, you would use the traditional dot(.
) annotation, as shown in the following example
@Value("${appA.secret}") String name = "World";
The preceding code would sets the value of the name
variable to appAsecret
.
1.0.0.BUILD-SNAPSHOT
This project provides Netflix OSS integrations for Spring Boot apps through autoconfiguration and binding to the Spring Environment and other Spring programming model idioms. With a few simple annotations you can quickly enable and configure the common patterns inside your application and build large distributed systems with battle-tested Netflix components. The patterns provided include Service Discovery (Eureka), Circuit Breaker (Hystrix), Intelligent Routing (Zuul) and Client Side Load Balancing (Ribbon).
Service Discovery is one of the key tenets of a microservice-based architecture. Trying to hand-configure each client or some form of convention can be difficult to do and can be brittle. Eureka is the Netflix Service Discovery Server and Client. The server can be configured and deployed to be highly available, with each server replicating state about the registered services to the others.
To include the Eureka Client in your project, use the starter with a group ID of org.springframework.cloud
and an artifact ID of spring-cloud-starter-netflix-eureka-client
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
When a client registers with Eureka, it provides meta-data about itself — such as host, port, health indicator URL, home page, and other details. Eureka receives heartbeat messages from each instance belonging to a service. If the heartbeat fails over a configurable timetable, the instance is normally removed from the registry.
The following example shows a minimal Eureka client application:
@SpringBootApplication @RestController public class Application { @RequestMapping("/") public String home() { return "Hello world"; } public static void main(String[] args) { new SpringApplicationBuilder(Application.class).web(true).run(args); } }
Note that the preceding example shows a normal Spring Boot application.
By having spring-cloud-starter-netflix-eureka-client
on the classpath, your application automatically registers with the Eureka Server. Configuration is required to locate the Eureka server, as shown in the following example:
application.yml.
eureka: client: serviceUrl: defaultZone: http://localhost:8761/eureka/
In the preceding example, "defaultZone" is a magic string fallback value that provides the service URL for any client that does not express a preference (in other words, it is a useful default).
The default application name (that is, the service ID), virtual host, and non-secure port (taken from the Environment
) are ${spring.application.name}
, ${spring.application.name}
and ${server.port}
, respectively.
Having spring-cloud-starter-netflix-eureka-client
on the classpath makes the app into both a Eureka “instance” (that is, it registers itself) and a “client” (it can query the registry to locate other services).
The instance behaviour is driven by eureka.instance.*
configuration keys, but the defaults are fine if you ensure that your application has a value for spring.application.name
(this is the default for the Eureka service ID or VIP).
See EurekaInstanceConfigBean and EurekaClientConfigBean for more details on the configurable options.
To disable the Eureka Discovery Client, you can set eureka.client.enabled
to false
. Eureka Discovery Client will also be disabled when spring.cloud.discovery.enabled
is set to false
.
HTTP basic authentication is automatically added to your eureka client if one of the eureka.client.serviceUrl.defaultZone
URLs has credentials embedded in it (curl style, as follows: http://user:password@localhost:8761/eureka
).
For more complex needs, you can create a @Bean
of type DiscoveryClientOptionalArgs
and inject ClientFilter
instances into it, all of which is applied to the calls from the client to the server.
![]() | Note |
---|---|
Because of a limitation in Eureka, it is not possible to support per-server basic auth credentials, so only the first set that are found is used. |
The status page and health indicators for a Eureka instance default to /info
and /health
respectively, which are the default locations of useful endpoints in a Spring Boot Actuator application.
You need to change these, even for an Actuator application if you use a non-default context path or servlet path (such as server.servletPath=/custom
). The following example shows the default values for the two settings:
application.yml.
eureka: instance: statusPageUrlPath: ${server.servletPath}/info healthCheckUrlPath: ${server.servletPath}/health
These links show up in the metadata that is consumed by clients and are used in some scenarios to decide whether to send requests to your application, so it is helpful if they are accurate.
![]() | Note |
---|---|
In Dalston it was also required to set the status and health check URLs when changing that management context path. This requirement was removed beginning in Edgware. |
If your app wants to be contacted over HTTPS, you can set two flags in the EurekaInstanceConfig
:
eureka.instance.[nonSecurePortEnabled]=[false]
eureka.instance.[securePortEnabled]=[true]
Doing so makes Eureka publish instance information that shows an explicit preference for secure communication.
The Spring Cloud DiscoveryClient
always returns a URI starting with https
for a service configured this way.
Similarly, when a service is configured this way, the Eureka (native) instance information has a secure health check URL.
Because of the way Eureka works internally, it still publishes a non-secure URL for the status and home pages unless you also override those explicitly. You can use placeholders to configure the eureka instance URLs, as shown in the following example:
application.yml.
eureka: instance: statusPageUrl: https://${eureka.hostname}/info healthCheckUrl: https://${eureka.hostname}/health homePageUrl: https://${eureka.hostname}/
(Note that ${eureka.hostname}
is a native placeholder only available
in later versions of Eureka. You could achieve the same thing with
Spring placeholders as well — for example, by using ${eureka.instance.hostName}
.)
![]() | Note |
---|---|
If your application runs behind a proxy, and the SSL termination is in the proxy (for example, if you run in Cloud Foundry or other platforms as a service), then you need to ensure that the proxy “forwarded” headers are intercepted and handled by the application. If the Tomcat container embedded in a Spring Boot application has explicit configuration for the 'X-Forwarded-\*` headers, this happens automatically. The links rendered by your app to itself being wrong (the wrong host, port, or protocol) is a sign that you got this configuration wrong. |
By default, Eureka uses the client heartbeat to determine if a client is up. Unless specified otherwise, the Discovery Client does not propagate the current health check status of the application, per the Spring Boot Actuator. Consequently, after successful registration, Eureka always announces that the application is in 'UP' state. This behavior can be altered by enabling Eureka health checks, which results in propagating application status to Eureka. As a consequence, every other application does not send traffic to applications in states other then 'UP'. The following example shows how to enable health checks for the client:
application.yml.
eureka: client: healthcheck: enabled: true
![]() | Warning |
---|---|
|
If you require more control over the health checks, consider implementing your own com.netflix.appinfo.HealthCheckHandler
.
It is worth spending a bit of time understanding how the Eureka metadata works, so you can use it in a way that makes sense in your platform.
There is standard metadata for information such as hostname, IP address, port numbers, the status page, and health check.
These are published in the service registry and used by clients to contact the services in a straightforward way.
Additional metadata can be added to the instance registration in the eureka.instance.metadataMap
, and this metadata is accessible in the remote clients.
In general, additional metadata does not change the behavior of the client, unless the client is made aware of the meaning of the metadata.
There are a couple of special cases, described later in this document, where Spring Cloud already assigns meaning to the metadata map.
Cloud Foundry has a global router so that all instances of the same app have the same hostname (other PaaS solutions with a similar architecture have the same arrangement).
This is not necessarily a barrier to using Eureka.
However, if you use the router (recommended or even mandatory, depending on the way your platform was set up), you need to explicitly set the hostname and port numbers (secure or non-secure) so that they use the router.
You might also want to use instance metadata so that you can distinguish between the instances on the client (for example, in a custom load balancer).
By default, the eureka.instance.instanceId
is vcap.application.instance_id
, as shown in the following example:
application.yml.
eureka: instance: hostname: ${vcap.application.uris[0]} nonSecurePort: 80
Depending on the way the security rules are set up in your Cloud Foundry instance, you might be able to register and use the IP address of the host VM for direct service-to-service calls. This feature is not yet available on Pivotal Web Services (PWS).
If the application is planned to be deployed to an AWS cloud, the Eureka instance must be configured to be AWS-aware. You can do so by customizing the EurekaInstanceConfigBean as follows:
@Bean @Profile("!default") public EurekaInstanceConfigBean eurekaInstanceConfig(InetUtils inetUtils) { EurekaInstanceConfigBean b = new EurekaInstanceConfigBean(inetUtils); AmazonInfo info = AmazonInfo.Builder.newBuilder().autoBuild("eureka"); b.setDataCenterInfo(info); return b; }
A vanilla Netflix Eureka instance is registered with an ID that is equal to its host name (that is, there is only one service per host). Spring Cloud Eureka provides a sensible default, which is defined as follows:
${spring.cloud.client.hostname}:${spring.application.name}:${spring.application.instance_id:${server.port}}}
An example is myhost:myappname:8080
.
By using Spring Cloud, you can override this value by providing a unique identifier in eureka.instance.instanceId
, as shown in the following example:
application.yml.
eureka: instance: instanceId: ${spring.application.name}:${vcap.application.instance_id:${spring.application.instance_id:${random.value}}}
With the metadata shown in the preceding example and multiple service instances deployed on localhost, the random value is inserted there to make the instance unique.
In Cloud Foundry, the vcap.application.instance_id
is populated automatically in a Spring Boot application, so the random value is not needed.
Once you have an application that is a discovery client, you can use it to discover service instances from the Eureka Server.
One way to do so is to use the native com.netflix.discovery.EurekaClient
(as opposed to the Spring Cloud DiscoveryClient
), as shown in the following example:
@Autowired private EurekaClient discoveryClient; public String serviceUrl() { InstanceInfo instance = discoveryClient.getNextServerFromEureka("STORES", false); return instance.getHomePageUrl(); }
![]() | Tip |
---|---|
Do not use the |
By default, EurekaClient uses Jersey for HTTP communication.
If you wish to avoid dependencies from Jersey, you can exclude it from your dependencies.
Spring Cloud auto-configures a transport client based on Spring RestTemplate
.
The following example shows Jersey being excluded:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> <exclusions> <exclusion> <groupId>com.sun.jersey</groupId> <artifactId>jersey-client</artifactId> </exclusion> <exclusion> <groupId>com.sun.jersey</groupId> <artifactId>jersey-core</artifactId> </exclusion> <exclusion> <groupId>com.sun.jersey.contribs</groupId> <artifactId>jersey-apache-client4</artifactId> </exclusion> </exclusions> </dependency>
You need not use the raw Netflix EurekaClient
.
Also, it is usually more convenient to use it behind a wrapper of some sort.
Spring Cloud has support for Feign (a REST client builder) and Spring RestTemplate
through the logical Eureka service identifiers (VIPs) instead of physical URLs.
To configure Ribbon with a fixed list of physical servers, you can set <client>.ribbon.listOfServers
to a comma-separated list of physical addresses (or hostnames), where <client>
is the ID of the client.
You can also use the org.springframework.cloud.client.discovery.DiscoveryClient
, which provides a simple API (not specific to Netflix) for discovery clients, as shown in the following example:
@Autowired private DiscoveryClient discoveryClient; public String serviceUrl() { List<ServiceInstance> list = discoveryClient.getInstances("STORES"); if (list != null && list.size() > 0 ) { return list.get(0).getUri(); } return null; }
Being an instance also involves a periodic heartbeat to the registry
(through the client’s serviceUrl
) with a default duration of 30 seconds.
A service is not available for discovery by clients until the instance, the server, and the client all have the same metadata in their local
cache (so it could take 3 heartbeats).
You can change the period by setting eureka.instance.leaseRenewalIntervalInSeconds
.
Setting it to a value of less than 30 speeds up the process of getting clients connected to other services.
In production, it is probably better to stick with the default, because of internal computations in the server that make assumptions about the lease renewal period.
If you have deployed Eureka clients to multiple zones, you may prefer that those clients use services within the same zone before trying services in another zone. To set that up, you need to configure your Eureka clients correctly.
First, you need to make sure you have Eureka servers deployed to each zone and that they are peers of each other. See the section on zones and regions for more information.
Next, you need to tell Eureka which zone your service is in.
You can do so by using the metadataMap
property.
For example, if service 1
is deployed to both zone 1
and zone 2
, you need to set the following Eureka properties in service 1
:
Service 1 in Zone 1
eureka.instance.metadataMap.zone = zone1 eureka.client.preferSameZoneEureka = true
Service 1 in Zone 2
eureka.instance.metadataMap.zone = zone2 eureka.client.preferSameZoneEureka = true
This section describes how to set up a Eureka server.
To include Eureka Server in your project, use the starter with a group ID of org.springframework.cloud
and an artifact ID of spring-cloud-starter-netflix-eureka-server
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
![]() | Note |
---|---|
If your project already uses Thymeleaf as its template engine, the Freemarker templates of the Eureka server may not be loaded correctly. In this case it is necessary to configure the template loader manually: |
application.yml.
spring: freemarker: template-loader-path: classpath:/templates/ prefer-file-system-access: false
The following example shows a minimal Eureka server:
@SpringBootApplication @EnableEurekaServer public class Application { public static void main(String[] args) { new SpringApplicationBuilder(Application.class).web(true).run(args); } }
The server has a home page with a UI and HTTP API endpoints for the normal Eureka functionality under /eureka/*
.
The following links have some Eureka background reading: flux capacitor and google group discussion.
![]() | Tip |
---|---|
Due to Gradle’s dependency resolution rules and the lack of a parent bom feature, depending on build.gradle. buildscript { dependencies { classpath("org.springframework.boot:spring-boot-gradle-plugin:{spring-boot-docs-version}") } } apply plugin: "spring-boot" dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:{spring-cloud-version}" } }
|
The Eureka server does not have a back end store, but the service instances in the registry all have to send heartbeats to keep their registrations up to date (so this can be done in memory). Clients also have an in-memory cache of Eureka registrations (so they do not have to go to the registry for every request to a service).
By default, every Eureka server is also a Eureka client and requires (at least one) service URL to locate a peer. If you do not provide it, the service runs and works, but it fills your logs with a lot of noise about not being able to register with the peer.
See also below for details of Ribbon support on the client side for Zones and Regions.
The combination of the two caches (client and server) and the heartbeats make a standalone Eureka server fairly resilient to failure, as long as there is some sort of monitor or elastic runtime (such as Cloud Foundry) keeping it alive. In standalone mode, you might prefer to switch off the client side behavior so that it does not keep trying and failing to reach its peers. The following example shows how to switch off the client-side behavior:
application.yml (Standalone Eureka Server).
server: port: 8761 eureka: instance: hostname: localhost client: registerWithEureka: false fetchRegistry: false serviceUrl: defaultZone: http://${eureka.instance.hostname}:${server.port}/eureka/
Notice that the serviceUrl
is pointing to the same host as the local instance.
Eureka can be made even more resilient and available by running multiple instances and asking them to register with each other.
In fact, this is the default behavior, so all you need to do to make it work is add a valid serviceUrl
to a peer, as shown in the following example:
application.yml (Two Peer Aware Eureka Servers).
--- spring: profiles: peer1 eureka: instance: hostname: peer1 client: serviceUrl: defaultZone: http://peer2/eureka/ --- spring: profiles: peer2 eureka: instance: hostname: peer2 client: serviceUrl: defaultZone: http://peer1/eureka/
In the preceding example, we have a YAML file that can be used to run the same server on two hosts (peer1
and peer2
) by running it in different Spring profiles.
You could use this configuration to test the peer awareness on a single host (there is not much value in doing that in production) by manipulating /etc/hosts
to resolve the host names.
In fact, the eureka.instance.hostname
is not needed if you are running on a machine that knows its own hostname (by default, it is looked up by using java.net.InetAddress
).
You can add multiple peers to a system, and, as long as they are all connected to each other by at least one edge, they synchronize the registrations amongst themselves. If the peers are physically separated (inside a data center or between multiple data centers), then the system can, in principle, survive “split-brain” type failures. You can add multiple peers to a system, and as long as they are all directly connected to each other, they will synchronize the registrations amongst themselves.
application.yml (Three Peer Aware Eureka Servers).
eureka: client: serviceUrl: defaultZone: http://peer1/eureka/,http://peer2/eureka/,http://peer3/eureka/ --- spring: profiles: peer1 eureka: instance: hostname: peer1 --- spring: profiles: peer2 eureka: instance: hostname: peer2 --- spring: profiles: peer3 eureka: instance: hostname: peer3
In some cases, it is preferable for Eureka to advertise the IP addresses of services rather than the hostname.
Set eureka.instance.preferIpAddress
to true
and, when the application registers with eureka, it uses its IP address rather than its hostname.
![]() | Tip |
---|---|
If the hostname cannot be determined by Java, then the IP address is sent to Eureka.
Only explict way of setting the hostname is by setting |
You can secure your Eureka server simply by adding Spring Security to your
server’s classpath via spring-boot-starter-security
. By default when Spring Security is on the classpath it will require that
a valid CSRF token be sent with every request to the app. Eureka clients will not generally possess a valid
cross site request forgery (CSRF) token you will need to disable this requirement for the /eureka/**
endpoints.
For example:
@EnableWebSecurity class WebSecurityConfig extends WebSecurityConfigurerAdapter { @Override protected void configure(HttpSecurity http) throws Exception { http.csrf().ignoringAntMatchers("/eureka/**"); super.configure(http); } }
For more information on CSRF see the Spring Security documentation.
A demo Eureka Server can be found in the Spring Cloud Samples repo.
The JAXB modules which the Eureka server depends upon were removed in JDK 11. If you intend to use JDK 11 when running a Eureka server you must include these dependencies in your POM or Gradle file.
<dependency> <groupId>org.glassfish.jaxb</groupId> <artifactId>jaxb-runtime</artifactId> </dependency>
Netflix has created a library called Hystrix that implements the circuit breaker pattern. In a microservice architecture, it is common to have multiple layers of service calls, as shown in the following example:
A service failure in the lower level of services can cause cascading failure all the way up to the user.
When calls to a particular service exceed circuitBreaker.requestVolumeThreshold
(default: 20 requests) and the failure percentage is greater than circuitBreaker.errorThresholdPercentage
(default: >50%) in a rolling window defined by metrics.rollingStats.timeInMilliseconds
(default: 10 seconds), the circuit opens and the call is not made.
In cases of error and an open circuit, a fallback can be provided by the developer.
Having an open circuit stops cascading failures and allows overwhelmed or failing services time to recover. The fallback can be another Hystrix protected call, static data, or a sensible empty value. Fallbacks may be chained so that the first fallback makes some other business call, which in turn falls back to static data.
To include Hystrix in your project, use the starter with a group ID of org.springframework.cloud
and a artifact ID of spring-cloud-starter-netflix-hystrix
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
The following example shows a minimal Eureka server with a Hystrix circuit breaker:
@SpringBootApplication @EnableCircuitBreaker public class Application { public static void main(String[] args) { new SpringApplicationBuilder(Application.class).web(true).run(args); } } @Component public class StoreIntegration { @HystrixCommand(fallbackMethod = "defaultStores") public Object getStores(Map<String, Object> parameters) { //do stuff that might fail } public Object defaultStores(Map<String, Object> parameters) { return /* something useful */; } }
The @HystrixCommand
is provided by a Netflix contrib library called “javanica”.
Spring Cloud automatically wraps Spring beans with that annotation in a proxy that is connected to the Hystrix circuit breaker.
The circuit breaker calculates when to open and close the circuit and what to do in case of a failure.
To configure the @HystrixCommand
you can use the commandProperties
attribute with a list of @HystrixProperty
annotations. See
here
for more details. See the Hystrix wiki
for details on the properties available.
If you want some thread local context to propagate into a @HystrixCommand
, the default declaration does not work, because it executes the command in a thread pool (in case of timeouts).
You can switch Hystrix to use the same thread as the caller through configuration or directly in the annotation, by asking it to use a different “Isolation Strategy”.
The following example demonstrates setting the thread in the annotation:
@HystrixCommand(fallbackMethod = "stubMyService",
commandProperties = {
@HystrixProperty(name="execution.isolation.strategy", value="SEMAPHORE")
}
)
...
The same thing applies if you are using @SessionScope
or @RequestScope
.
If you encounter a runtime exception that says it cannot find the scoped context, you need to use the same thread.
You also have the option to set the hystrix.shareSecurityContext
property to true
.
Doing so auto-configures a Hystrix concurrency strategy plugin hook to transfer the SecurityContext
from your main thread to the one used by the Hystrix command.
Hystrix does not let multiple Hystrix concurrency strategy be registered so an extension mechanism is available by declaring your own HystrixConcurrencyStrategy
as a Spring bean.
Spring Cloud looks for your implementation within the Spring context and wrap it inside its own plugin.
The state of the connected circuit breakers are also exposed in the /health
endpoint of the calling application, as shown in the following example:
{ "hystrix": { "openCircuitBreakers": [ "StoreIntegration::getStoresByLocationLink" ], "status": "CIRCUIT_OPEN" }, "status": "UP" }
To enable the Hystrix metrics stream, include a dependency on spring-boot-starter-actuator
and set
management.endpoints.web.exposure.include: hystrix.stream
.
Doing so exposes the /actuator/hystrix.stream
as a management endpoint, as shown in the following example:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency>
One of the main benefits of Hystrix is the set of metrics it gathers about each HystrixCommand. The Hystrix Dashboard displays the health of each circuit breaker in an efficient manner.
When using Hystrix commands that wrap Ribbon clients you want to make sure your Hystrix timeout is configured to be longer than the configured Ribbon timeout, including any potential retries that might be made. For example, if your Ribbon connection timeout is one second and the Ribbon client might retry the request three times, than your Hystrix timeout should be slightly more than three seconds.
To include the Hystrix Dashboard in your project, use the starter with a group ID of org.springframework.cloud
and an artifact ID of spring-cloud-starter-netflix-hystrix-dashboard
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
To run the Hystrix Dashboard, annotate your Spring Boot main class with @EnableHystrixDashboard
.
Then visit /hystrix
and point the dashboard to an individual instance’s /hystrix.stream
endpoint in a Hystrix client application.
![]() | Note |
---|---|
When connecting to a |
Looking at an individual instance’s Hystrix data is not very useful in terms of the overall health of the system. Turbine is an application that aggregates all of the relevant /hystrix.stream
endpoints into a combined /turbine.stream
for use in the Hystrix Dashboard.
Individual instances are located through Eureka.
Running Turbine requires annotating your main class with the @EnableTurbine
annotation (for example, by using spring-cloud-starter-netflix-turbine to set up the classpath).
All of the documented configuration properties from the Turbine 1 wiki apply.
The only difference is that the turbine.instanceUrlSuffix
does not need the port prepended, as this is handled automatically unless turbine.instanceInsertPort=false
.
![]() | Note |
---|---|
By default, Turbine looks for the |
eureka: instance: metadata-map: management.port: ${management.port:8081}
The turbine.appConfig
configuration key is a list of Eureka serviceIds that turbine uses to lookup instances.
The turbine stream is then used in the Hystrix dashboard with a URL similar to the following:
http://my.turbine.server:8080/turbine.stream?cluster=CLUSTERNAME
The cluster parameter can be omitted if the name is default
.
The cluster
parameter must match an entry in turbine.aggregator.clusterConfig
.
Values returned from Eureka are upper-case. Consequently, the following example works if there is an application called customers
registered with Eureka:
turbine: aggregator: clusterConfig: CUSTOMERS appConfig: customers
If you need to customize which cluster names should be used by Turbine (because you do not want to store cluster names in
turbine.aggregator.clusterConfig
configuration), provide a bean of type TurbineClustersProvider
.
The clusterName
can be customized by a SPEL expression in turbine.clusterNameExpression
with root as an instance of InstanceInfo
.
The default value is appName
, which means that the Eureka serviceId
becomes the cluster key (that is, the InstanceInfo
for customers has an appName
of CUSTOMERS
).
A different example is turbine.clusterNameExpression=aSGName
, which gets the cluster name from the AWS ASG name.
The following listing shows another example:
turbine: aggregator: clusterConfig: SYSTEM,USER appConfig: customers,stores,ui,admin clusterNameExpression: metadata['cluster']
In the preceding example, the cluster name from four services is pulled from their metadata map and is expected to have values that include SYSTEM
and USER
.
To use the “default” cluster for all apps, you need a string literal expression (with single quotes and escaped with double quotes if it is in YAML as well):
turbine: appConfig: customers,stores clusterNameExpression: "'default'"
Spring Cloud provides a spring-cloud-starter-netflix-turbine
that has all the dependencies you need to get a Turbine server running. To add Turbine, create a Spring Boot application and annotate it with @EnableTurbine
.
![]() | Note |
---|---|
By default, Spring Cloud lets Turbine use the host and port to allow multiple processes per host, per cluster.
If you want the native Netflix behavior built into Turbine to not allow multiple processes per host, per cluster (the key to the instance ID is the hostname), set |
In some situations it might be useful for other applications to know what custers have been configured
in Turbine. To support this you can use the /clusters
endpoint which will return a JSON array of
all the configured clusters.
GET /clusters.
[ { "name": "RACES", "link": "http://localhost:8383/turbine.stream?cluster=RACES" }, { "name": "WEB", "link": "http://localhost:8383/turbine.stream?cluster=WEB" } ]
This endpoint can be disabled by setting turbine.endpoints.clusters.enabled
to false
.
In some environments (such as in a PaaS setting), the classic Turbine model of pulling metrics from all the distributed Hystrix commands does not work.
In that case, you might want to have your Hystrix commands push metrics to Turbine. Spring Cloud enables that with messaging.
To do so on the client, add a dependency to spring-cloud-netflix-hystrix-stream
and the spring-cloud-starter-stream-*
of your choice.
See the Spring Cloud Stream documentation for details on the brokers and how to configure the client credentials. It should work out of the box for a local broker.
On the server side, create a Spring Boot application and annotate it with @EnableTurbineStream
.
The Turbine Stream server requires the use of Spring Webflux, therefore spring-boot-starter-webflux
needs to be included in your project.
By default spring-boot-starter-webflux
is included when adding spring-cloud-starter-netflix-turbine-stream
to your application.
You can then point the Hystrix Dashboard to the Turbine Stream Server instead of individual Hystrix streams.
If Turbine Stream is running on port 8989 on myhost, then put http://myhost:8989
in the stream input field in the Hystrix Dashboard.
Circuits are prefixed by their respective serviceId
, followed by a dot (.
), and then the circuit name.
Spring Cloud provides a spring-cloud-starter-netflix-turbine-stream
that has all the dependencies you need to get a Turbine Stream server running.
You can then add the Stream binder of your choice — such as spring-cloud-starter-stream-rabbit
.
Turbine Stream server also supports the cluster
parameter.
Unlike Turbine server, Turbine Stream uses eureka serviceIds as cluster names and these are not configurable.
If Turbine Stream server is running on port 8989 on my.turbine.server
and you have two eureka serviceIds customers
and products
in your environment, the following URLs will be available on your Turbine Stream server. default
and empty cluster name will provide all metrics that Turbine Stream server receives.
http://my.turbine.sever:8989/turbine.stream?cluster=customers http://my.turbine.sever:8989/turbine.stream?cluster=products http://my.turbine.sever:8989/turbine.stream?cluster=default http://my.turbine.sever:8989/turbine.stream
So, you can use eureka serviceIds as cluster names for your Turbine dashboard (or any compatible dashboard).
You don’t need to configure any properties like turbine.appConfig
, turbine.clusterNameExpression
and turbine.aggregator.clusterConfig
for your Turbine Stream server.
![]() | Note |
---|---|
Turbine Stream server gathers all metrics from the configured input channel with Spring Cloud Stream. It means that it doesn’t gather Hystrix metrics actively from each instance. It just can provide metrics that were already gathered into the input channel by each instance. |
Ribbon is a client-side load balancer that gives you a lot of control over the behavior of HTTP and TCP clients.
Feign already uses Ribbon, so, if you use @FeignClient
, this section also applies.
A central concept in Ribbon is that of the named client.
Each load balancer is part of an ensemble of components that work together to contact a remote server on demand, and the ensemble has a name that you give it as an application developer (for example, by using the @FeignClient
annotation).
On demand, Spring Cloud creates a new ensemble as an ApplicationContext
for each named client by using
RibbonClientConfiguration
.
This contains (amongst other things) an ILoadBalancer
, a RestClient
, and a ServerListFilter
.
To include Ribbon in your project, use the starter with a group ID of org.springframework.cloud
and an artifact ID of spring-cloud-starter-netflix-ribbon
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
You can configure some bits of a Ribbon client by using external properties in <client>.ribbon.*
, which is similar to using the Netflix APIs natively, except that you can use Spring Boot configuration files.
The native options can be inspected as static fields in CommonClientConfigKey
(part of ribbon-core).
Spring Cloud also lets you take full control of the client by declaring additional configuration (on top of the RibbonClientConfiguration
) using @RibbonClient
, as shown in the following example:
@Configuration @RibbonClient(name = "custom", configuration = CustomConfiguration.class) public class TestConfiguration { }
In this case, the client is composed from the components already in RibbonClientConfiguration
, together with any in CustomConfiguration
(where the latter generally overrides the former).
![]() | Warning |
---|---|
The |
The following table shows the beans that Spring Cloud Netflix provides by default for Ribbon:
Bean Type | Bean Name | Class Name |
---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Creating a bean of one of those type and placing it in a @RibbonClient
configuration (such as FooConfiguration
above) lets you override each one of the beans described, as shown in the following example:
@Configuration protected static class FooConfiguration { @Bean public ZonePreferenceServerListFilter serverListFilter() { ZonePreferenceServerListFilter filter = new ZonePreferenceServerListFilter(); filter.setZone("myTestZone"); return filter; } @Bean public IPing ribbonPing() { return new PingUrl(); } }
The include statement in the preceding example replaces NoOpPing
with PingUrl
and provides a custom serverListFilter
.
A default configuration can be provided for all Ribbon Clients by using the @RibbonClients
annotation and registering a default configuration, as shown in the following example:
@RibbonClients(defaultConfiguration = DefaultRibbonConfig.class) public class RibbonClientDefaultConfigurationTestsConfig { public static class BazServiceList extends ConfigurationBasedServerList { public BazServiceList(IClientConfig config) { super.initWithNiwsConfig(config); } } } @Configuration class DefaultRibbonConfig { @Bean public IRule ribbonRule() { return new BestAvailableRule(); } @Bean public IPing ribbonPing() { return new PingUrl(); } @Bean public ServerList<Server> ribbonServerList(IClientConfig config) { return new RibbonClientDefaultConfigurationTestsConfig.BazServiceList(config); } @Bean public ServerListSubsetFilter serverListFilter() { ServerListSubsetFilter filter = new ServerListSubsetFilter(); return filter; } }
Starting with version 1.2.0, Spring Cloud Netflix now supports customizing Ribbon clients by setting properties to be compatible with the Ribbon documentation.
This lets you change behavior at start up time in different environments.
The following list shows the supported properties>:
<clientName>.ribbon.NFLoadBalancerClassName
: Should implement ILoadBalancer
<clientName>.ribbon.NFLoadBalancerRuleClassName
: Should implement IRule
<clientName>.ribbon.NFLoadBalancerPingClassName
: Should implement IPing
<clientName>.ribbon.NIWSServerListClassName
: Should implement ServerList
<clientName>.ribbon.NIWSServerListFilterClassName
: Should implement ServerListFilter
![]() | Note |
---|---|
Classes defined in these properties have precedence over beans defined by using |
To set the IRule
for a service name called users
, you could set the following properties:
application.yml.
users: ribbon: NIWSServerListClassName: com.netflix.loadbalancer.ConfigurationBasedServerList NFLoadBalancerRuleClassName: com.netflix.loadbalancer.WeightedResponseTimeRule
See the Ribbon documentation for implementations provided by Ribbon.
When Eureka is used in conjunction with Ribbon (that is, both are on the classpath), the ribbonServerList
is overridden with an extension of DiscoveryEnabledNIWSServerList
, which populates the list of servers from Eureka.
It also replaces the IPing
interface with NIWSDiscoveryPing
, which delegates to Eureka to determine if a server is up.
The ServerList
that is installed by default is a DomainExtractingServerList
. Its purpose is to make metadata available to the load balancer without using AWS AMI metadata (which is what Netflix relies on).
By default, the server list is constructed with “zone” information, as provided in the instance metadata (so, on the remote clients, set eureka.instance.metadataMap.zone
).
If that is missing and if the approximateZoneFromHostname
flag is set, it can use the domain name from the server hostname as a proxy for the zone.
Once the zone information is available, it can be used in a ServerListFilter
.
By default, it is used to locate a server in the same zone as the client, because the default is a ZonePreferenceServerListFilter
.
By default, the zone of the client is determined in the same way as the remote instances (that is, through eureka.instance.metadataMap.zone
).
![]() | Note |
---|---|
The orthodox “archaius” way to set the client zone is through a configuration property called "@zone". If it is available, Spring Cloud uses that in preference to all other settings (note that the key must be quoted in YAML configuration). |
![]() | Note |
---|---|
If there is no other source of zone data, then a guess is made, based on the client configuration (as opposed to the instance configuration).
We take |
Eureka is a convenient way to abstract the discovery of remote servers so that you do not have to hard code their URLs in clients.
However, if you prefer not to use Eureka, Ribbon and Feign also work.
Suppose you have declared a @RibbonClient
for "stores", and Eureka is not in use (and not even on the classpath).
The Ribbon client defaults to a configured server list.
You can supply the configuration as follows:
application.yml.
stores: ribbon: listOfServers: example.com,google.com
Setting the ribbon.eureka.enabled
property to false
explicitly disables the use of Eureka in Ribbon, as shown in the following example:
application.yml.
ribbon: eureka: enabled: false
You can also use the LoadBalancerClient
directly, as shown in the following example:
public class MyClass { @Autowired private LoadBalancerClient loadBalancer; public void doStuff() { ServiceInstance instance = loadBalancer.choose("stores"); URI storesUri = URI.create(String.format("http://%s:%s", instance.getHost(), instance.getPort())); // ... do something with the URI } }
Each Ribbon named client has a corresponding child application Context that Spring Cloud maintains. This application context is lazily loaded on the first request to the named client. This lazy loading behavior can be changed to instead eagerly load these child application contexts at startup, by specifying the names of the Ribbon clients, as shown in the following example:
application.yml.
ribbon: eager-load: enabled: true clients: client1, client2, client3
If you change zuul.ribbonIsolationStrategy
to THREAD
, the thread isolation strategy for Hystrix is used for all routes.
In that case, the HystrixThreadPoolKey
is set to RibbonCommand
as the default.
It means that HystrixCommands for all routes are executed in the same Hystrix thread pool.
This behavior can be changed with the following configuration:
application.yml.
zuul: threadPool: useSeparateThreadPools: true
The preceding example results in HystrixCommands being executed in the Hystrix thread pool for each route.
In this case, the default HystrixThreadPoolKey
is the same as the service ID for each route.
To add a prefix to HystrixThreadPoolKey
, set zuul.threadPool.threadPoolKeyPrefix
to the value that you want to add, as shown in the following example:
application.yml.
zuul: threadPool: useSeparateThreadPools: true threadPoolKeyPrefix: zuulgw
If you need to provide your own IRule
implementation to handle a special routing requirement like a “canary” test, pass some information to the choose
method of IRule
.
com.netflix.loadbalancer.IRule.java.
public interface IRule{ public Server choose(Object key); :
You can provide some information that is used by your IRule
implementation to choose a target server, as shown in the following example:
RequestContext.getCurrentContext() .set(FilterConstants.LOAD_BALANCER_KEY, "canary-test");
If you put any object into the RequestContext
with a key of FilterConstants.LOAD_BALANCER_KEY
, it is passed to the choose
method of the IRule
implementation.
The code shown in the preceding example must be executed before RibbonRoutingFilter
is executed.
Zuul’s pre filter is the best place to do that.
You can access HTTP headers and query parameters through the RequestContext
in pre filter, so it can be used to determine the LOAD_BALANCER_KEY
that is passed to Ribbon.
If you do not put any value with LOAD_BALANCER_KEY
in RequestContext
, null is passed as a parameter of the choose
method.
Archaius is the Netflix client-side configuration library. It is the library used by all of the Netflix OSS components for configuration. Archaius is an extension of the Apache Commons Configuration project. It allows updates to configuration by either polling a source for changes or by letting a source push changes to the client. Archaius uses Dynamic<Type>Property classes as handles to properties, as shown in the following example:
Archaius Example.
class ArchaiusTest { DynamicStringProperty myprop = DynamicPropertyFactory .getInstance() .getStringProperty("my.prop"); void doSomething() { OtherClass.someMethod(myprop.get()); } }
Archaius has its own set of configuration files and loading priorities. Spring applications should generally not use Archaius directly, but the need to configure the Netflix tools natively remains. Spring Cloud has a Spring Environment Bridge so that Archaius can read properties from the Spring Environment. This bridge allows Spring Boot projects to use the normal configuration toolchain while letting them configure the Netflix tools as documented (for the most part).
Routing is an integral part of a microservice architecture.
For example, /
may be mapped to your web application, /api/users
is mapped to the user service and /api/shop
is mapped to the shop service.
Zuul is a JVM-based router and server-side load balancer from Netflix.
Netflix uses Zuul for the following:
Zuul’s rule engine lets rules and filters be written in essentially any JVM language, with built-in support for Java and Groovy.
![]() | Note |
---|---|
The configuration property |
![]() | Note |
---|---|
The default Hystrix isolation pattern ( |
To include Zuul in your project, use the starter with a group ID of org.springframework.cloud
and a artifact ID of spring-cloud-starter-netflix-zuul
.
See the Spring Cloud Project page for details on setting up your build system with the current Spring Cloud Release Train.
Spring Cloud has created an embedded Zuul proxy to ease the development of a common use case where a UI application wants to make proxy calls to one or more back end services. This feature is useful for a user interface to proxy to the back end services it requires, avoiding the need to manage CORS and authentication concerns independently for all the back ends.
To enable it, annotate a Spring Boot main class with @EnableZuulProxy
. Doing so causes local calls to be forwarded to the appropriate service.
By convention, a service with an ID of users
receives requests from the proxy located at /users
(with the prefix stripped).
The proxy uses Ribbon to locate an instance to which to forward through discovery.
All requests are executed in a hystrix command, so failures appear in Hystrix metrics.
Once the circuit is open, the proxy does not try to contact the service.
![]() | Note |
---|---|
the Zuul starter does not include a discovery client, so, for routes based on service IDs, you need to provide one of those on the classpath as well (Eureka is one choice). |
To skip having a service automatically added, set zuul.ignored-services
to a list of service ID patterns.
If a service matches a pattern that is ignored but is also included in the explicitly configured routes map, it is unignored, as shown in the following example:
application.yml.
zuul: ignoredServices: '*' routes: users: /myusers/**
In the preceding example, all services are ignored, except for users
.
To augment or change the proxy routes, you can add external configuration, as follows:
application.yml.
zuul: routes: users: /myusers/**
The preceding example means that HTTP calls to /myusers
get forwarded to the users
service (for example /myusers/101
is forwarded to /101
).
To get more fine-grained control over a route, you can specify the path and the serviceId independently, as follows:
application.yml.
zuul: routes: users: path: /myusers/** serviceId: users_service
The preceding example means that HTTP calls to /myusers
get forwarded to the users_service
service.
The route must have a path
that can be specified as an ant-style pattern, so /myusers/*
only matches one level, but /myusers/**
matches hierarchically.
The location of the back end can be specified as either a serviceId
(for a service from discovery) or a url
(for a physical location), as shown in the following example:
application.yml.
zuul: routes: users: path: /myusers/** url: http://example.com/users_service
These simple url-routes do not get executed as a HystrixCommand
, nor do they load-balance multiple URLs with Ribbon.
To achieve those goals, you can specify a serviceId
with a static list of servers, as follows:
application.yml.
zuul: routes: echo: path: /myusers/** serviceId: myusers-service stripPrefix: true hystrix: command: myusers-service: execution: isolation: thread: timeoutInMilliseconds: ... myusers-service: ribbon: NIWSServerListClassName: com.netflix.loadbalancer.ConfigurationBasedServerList listOfServers: http://example1.com,http://example2.com ConnectTimeout: 1000 ReadTimeout: 3000 MaxTotalHttpConnections: 500 MaxConnectionsPerHost: 100
Another method is specifiying a service-route and configuring a Ribbon client for the serviceId
(doing so requires disabling Eureka support in Ribbon — see above for more information), as shown in the following example:
application.yml.
zuul: routes: users: path: /myusers/** serviceId: users ribbon: eureka: enabled: false users: ribbon: listOfServers: example.com,google.com
You can provide a convention between serviceId
and routes by using regexmapper
.
It uses regular-expression named groups to extract variables from serviceId
and inject them into a route pattern, as shown in the following example:
ApplicationConfiguration.java.
@Bean public PatternServiceRouteMapper serviceRouteMapper() { return new PatternServiceRouteMapper( "(?<name>^.+)-(?<version>v.+$)", "${version}/${name}"); }
The preceding example means that a serviceId
of myusers-v1
is mapped to route /v1/myusers/**
.
Any regular expression is accepted, but all named groups must be present in both servicePattern
and routePattern
.
If servicePattern
does not match a serviceId
, the default behavior is used.
In the preceding example, a serviceId
of myusers
is mapped to the "/myusers/**" route (with no version detected).
This feature is disabled by default and only applies to discovered services.
To add a prefix to all mappings, set zuul.prefix
to a value, such as /api
.
By default, the proxy prefix is stripped from the request before the request is forwarded by (you can switch this behavior off with zuul.stripPrefix=false
).
You can also switch off the stripping of the service-specific prefix from individual routes, as shown in the following example:
application.yml.
zuul: routes: users: path: /myusers/** stripPrefix: false
![]() | Note |
---|---|
|
In the preceding example, requests to /myusers/101
are forwarded to /myusers/101
on the users
service.
The zuul.routes
entries actually bind to an object of type ZuulProperties
.
If you look at the properties of that object, you can see that it also has a retryable
flag.
Set that flag to true
to have the Ribbon client automatically retry failed requests.
You can also set that flag to true
when you need to modify the parameters of the retry operations that use the Ribbon client configuration.
By default, the X-Forwarded-Host
header is added to the forwarded requests.
To turn it off, set zuul.addProxyHeaders = false
.
By default, the prefix path is stripped, and the request to the back end picks up a X-Forwarded-Prefix
header (/myusers
in the examples shown earlier).
If you set a default route (/
), an application with @EnableZuulProxy
could act as a standalone server. For example, zuul.route.home: /
would route all traffic ("/**") to the "home" service.
If more fine-grained ignoring is needed, you can specify specific patterns to ignore. These patterns are evaluated at the start of the route location process, which means prefixes should be included in the pattern to warrant a match. Ignored patterns span all services and supersede any other route specification. The following example shows how to create ignored patterns:
application.yml.
zuul: ignoredPatterns: /**/admin/** routes: users: /myusers/**
The preceding example means that all calls (such as /myusers/101
) are forwarded to /101
on the users
service.
However, calls including /admin/
do not resolve.
![]() | Warning |
---|---|
If you need your routes to have their order preserved, you need to use a YAML file, as the ordering is lost when using a properties file. The following example shows such a YAML file: |
application.yml.
zuul: routes: users: path: /myusers/** legacy: path: /**
If you were to use a properties file, the legacy
path might end up in front of the users
path, rendering the users
path unreachable.
The default HTTP client used by Zuul is now backed by the Apache HTTP Client instead of the deprecated Ribbon RestClient
.
To use RestClient
or okhttp3.OkHttpClient
, set ribbon.restclient.enabled=true
or ribbon.okhttp.enabled=true
, respectively.
If you would like to customize the Apache HTTP client or the OK HTTP client, provide a bean of type ClosableHttpClient
or OkHttpClient
.
You can share headers between services in the same system, but you probably do not want sensitive headers leaking downstream into external servers. You can specify a list of ignored headers as part of the route configuration. Cookies play a special role, because they have well defined semantics in browsers, and they are always to be treated as sensitive. If the consumer of your proxy is a browser, then cookies for downstream services also cause problems for the user, because they all get jumbled up together (all downstream services look like they come from the same place).
If you are careful with the design of your services, (for example, if only one of the downstream services sets cookies), you might be able to let them flow from the back end all the way up to the caller.
Also, if your proxy sets cookies and all your back-end services are part of the same system, it can be natural to simply share them (and, for instance, use Spring Session to link them up to some shared state).
Other than that, any cookies that get set by downstream services are likely to be not useful to the caller, so it is recommended that you make (at least) Set-Cookie
and Cookie
into sensitive headers for routes that are not part of your domain.
Even for routes that are part of your domain, try to think carefully about what it means before letting cookies flow between them and the proxy.
The sensitive headers can be configured as a comma-separated list per route, as shown in the following example:
application.yml.
zuul: routes: users: path: /myusers/** sensitiveHeaders: Cookie,Set-Cookie,Authorization url: https://downstream
![]() | Note |
---|---|
This is the default value for |
The sensitiveHeaders
are a blacklist, and the default is not empty.
Consequently, to make Zuul send all headers (except the ignored
ones), you must explicitly set it to the empty list.
Doing so is necessary if you want to pass cookie or authorization headers to your back end. The following example shows how to use sensitiveHeaders
:
application.yml.
zuul: routes: users: path: /myusers/** sensitiveHeaders: url: https://downstream
You can also set sensitive headers, by setting zuul.sensitiveHeaders
.
If sensitiveHeaders
is set on a route, it overrides the global sensitiveHeaders
setting.
In addition to the route-sensitive headers, you can set a global value called zuul.ignoredHeaders
for values (both request and response) that should be discarded during interactions with downstream services.
By default, if Spring Security is not on the classpath, these are empty.
Otherwise, they are initialized to a set of well known “security” headers (for example, involving caching) as specified by Spring Security.
The assumption in this case is that the downstream services might add these headers, too, but we want the values from the proxy.
To not discard these well known security headers when Spring Security is on the classpath, you can set zuul.ignoreSecurityHeaders
to false
.
Doing so can be useful if you disabled the HTTP Security response headers in Spring Security and want the values provided by downstream services.
By default, if you use @EnableZuulProxy
with the Spring Boot Actuator, you enable two additional endpoints:
A GET to the routes endpoint at /routes
returns a list of the mapped routes:
GET /routes.
{ /stores/**: "http://localhost:8081" }
Additional route details can be requested by adding the ?format=details
query string to /routes
.
Doing so produces the following output:
GET /routes/details.
{ "/stores/**": { "id": "stores", "fullPath": "/stores/**", "location": "http://localhost:8081", "path": "/**", "prefix": "/stores", "retryable": false, "customSensitiveHeaders": false, "prefixStripped": true } }
A POST
to /routes
forces a refresh of the existing routes (for example, when there have been changes in the service catalog).
You can disable this endpoint by setting endpoints.routes.enabled
to false
.
![]() | Note |
---|---|
the routes should respond automatically to changes in the service catalog, but the |
A common pattern when migrating an existing application or API is to “strangle” old endpoints, slowly replacing them with different implementations. The Zuul proxy is a useful tool for this because you can use it to handle all traffic from the clients of the old endpoints but redirect some of the requests to new ones.
The following example shows the configuration details for a “strangle” scenario:
application.yml.
zuul: routes: first: path: /first/** url: http://first.example.com second: path: /second/** url: forward:/second third: path: /third/** url: forward:/3rd legacy: path: /** url: http://legacy.example.com
In the preceding example, we are strangle the “legacy” application, which is mapped to all requests that do not match one of the other patterns.
Paths in /first/**
have been extracted into a new service with an external URL.
Paths in /second/**
are forwarded so that they can be handled locally (for example, with a normal Spring @RequestMapping
).
Paths in /third/**
are also forwarded but with a different prefix (/third/foo
is forwarded to /3rd/foo
).
![]() | Note |
---|---|
The ignored patterns aren’t completely ignored, they just are not handled by the proxy (so they are also effectively forwarded locally). |
If you use @EnableZuulProxy
, you can use the proxy paths to upload files and it should work, so long as the files are small.
For large files there is an alternative path that bypasses the Spring DispatcherServlet
(to avoid multipart processing) in "/zuul/*".
In other words, if you have zuul.routes.customers=/customers/**
, then you can POST
large files to /zuul/customers/*
.
The servlet path is externalized via zuul.servletPath
.
If the proxy route takes you through a Ribbon load balancer, extremely large files also require elevated timeout settings, as shown in the following example:
application.yml.
hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds: 60000 ribbon: ConnectTimeout: 3000 ReadTimeout: 60000
Note that, for streaming to work with large files, you need to use chunked encoding in the request (which some browsers do not do by default), as shown in the following example:
$ curl -v -H "Transfer-Encoding: chunked" \ -F "[email protected]" localhost:9999/zuul/simple/file
When processing the incoming request, query params are decoded so that they can be available for possible modifications in Zuul filters.
They are then re-encoded the back end request is rebuilt in the route filters.
The result can be different than the original input if (for example) it was encoded with Javascript’s encodeURIComponent()
method.
While this causes no issues in most cases, some web servers can be picky with the encoding of complex query string.
To force the original encoding of the query string, it is possible to pass a special flag to ZuulProperties
so that the query string is taken as is with the HttpServletRequest::getQueryString
method, as shown in the following example:
application.yml.
zuul: forceOriginalQueryStringEncoding: true
![]() | Note |
---|---|
This special flag works only with |
When processing the incoming request, request URI is decoded before matching them to routes. The request URI is then re-encoded when the back end request is rebuilt in the route filters. This can cause some unexpected behavior if your URI includes the encoded "/" character.
To use the original request URI, it is possible to pass a special flag to 'ZuulProperties' so that the URI will be taken as is with the HttpServletRequest::getRequestURI
method, as shown in the following example:
application.yml.
zuul: decodeUrl: false
![]() | Note |
---|---|
If you are overriding request URI using |
If you use @EnableZuulServer
(instead of @EnableZuulProxy
), you can also run a Zuul server without proxying or selectively switch on parts of the proxying platform.
Any beans that you add to the application of type ZuulFilter
are installed automatically (as they are with @EnableZuulProxy
) but without any of the proxy filters being added automatically.
In that case, the routes into the Zuul server are still specified by configuring "zuul.routes.*", but there is no service discovery and no proxying. Consequently, the "serviceId" and "url" settings are ignored. The following example maps all paths in "/api/**" to the Zuul filter chain:
application.yml.
zuul: routes: api: /api/**
Zuul for Spring Cloud comes with a number of ZuulFilter
beans enabled by default in both proxy and server mode.
See the Zuul filters package for the list of filters that you can enable.
If you want to disable one, set zuul.<SimpleClassName>.<filterType>.disable=true
.
By convention, the package after filters
is the Zuul filter type.
For example to disable org.springframework.cloud.netflix.zuul.filters.post.SendResponseFilter
, set zuul.SendResponseFilter.post.disable=true
.
When a circuit for a given route in Zuul is tripped, you can provide a fallback response by creating a bean of type FallbackProvider
.
Within this bean, you need to specify the route ID the fallback is for and provide a ClientHttpResponse
to return as a fallback.
The following example shows a relatively simple FallbackProvider
implementation:
class MyFallbackProvider implements FallbackProvider { @Override public String getRoute() { return "customers"; } @Override public ClientHttpResponse fallbackResponse(String route, final Throwable cause) { if (cause instanceof HystrixTimeoutException) { return response(HttpStatus.GATEWAY_TIMEOUT); } else { return response(HttpStatus.INTERNAL_SERVER_ERROR); } } private ClientHttpResponse response(final HttpStatus status) { return new ClientHttpResponse() { @Override public HttpStatus getStatusCode() throws IOException { return status; } @Override public int getRawStatusCode() throws IOException { return status.value(); } @Override public String getStatusText() throws IOException { return status.getReasonPhrase(); } @Override public void close() { } @Override public InputStream getBody() throws IOException { return new ByteArrayInputStream("fallback".getBytes()); } @Override public HttpHeaders getHeaders() { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); return headers; } }; } }
The following example shows how the route configuration for the previous example might appear:
zuul: routes: customers: /customers/**
If you would like to provide a default fallback for all routes, you can create a bean of type FallbackProvider
and have the getRoute
method return *
or null
, as shown in the following example:
class MyFallbackProvider implements FallbackProvider { @Override public String getRoute() { return "*"; } @Override public ClientHttpResponse fallbackResponse(String route, Throwable throwable) { return new ClientHttpResponse() { @Override public HttpStatus getStatusCode() throws IOException { return HttpStatus.OK; } @Override public int getRawStatusCode() throws IOException { return 200; } @Override public String getStatusText() throws IOException { return "OK"; } @Override public void close() { } @Override public InputStream getBody() throws IOException { return new ByteArrayInputStream("fallback".getBytes()); } @Override public HttpHeaders getHeaders() { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); return headers; } }; } }
If you want to configure the socket timeouts and read timeouts for requests proxied through Zuul, you have two options, based on your configuration:
ribbon.ReadTimeout
and ribbon.SocketTimeout
Ribbon properties.If you have configured Zuul routes by specifying URLs, you need to use
zuul.host.connect-timeout-millis
and zuul.host.socket-timeout-millis
.
If Zuul is fronting a web application, you may need to re-write the Location
header when the web application redirects through a HTTP status code of 3XX
.
Otherwise, the browser redirects to the web application’s URL instead of the Zuul URL.
You can configure a LocationRewriteFilter
Zuul filter to re-write the Location
header to the Zuul’s URL.
It also adds back the stripped global and route-specific prefixes.
The following example adds a filter by using a Spring Configuration file:
import org.springframework.cloud.netflix.zuul.filters.post.LocationRewriteFilter; ... @Configuration @EnableZuulProxy public class ZuulConfig { @Bean public LocationRewriteFilter locationRewriteFilter() { return new LocationRewriteFilter(); } }
![]() | Caution |
---|---|
Use this filter carefully. The filter acts on the |
By default Zuul routes all Cross Origin requests (CORS) to the services. If you want instead Zuul to handle these requests it can be done by providing custom WebMvcConfigurer
bean:
@Bean public WebMvcConfigurer corsConfigurer() { return new WebMvcConfigurer() { public void addCorsMappings(CorsRegistry registry) { registry.addMapping("/path-1/**") .allowedOrigins("http://allowed-origin.com") .allowedMethods("GET", "POST"); } }; }
In the example above, we allow GET
and POST
methods from http://allowed-origin.com
to send cross-origin requests to the endpoints starting with path-1
.
You can apply CORS configuration to a specific path pattern or globally for the whole application, using /**
mapping.
You can customize properties: allowedOrigins
,allowedMethods
,allowedHeaders
,exposedHeaders
,allowCredentials
and maxAge
via this configuration.
Zuul will provide metrics under the Actuator metrics endpoint for any failures that might occur when routing requests.
These metrics can be viewed by hitting /actuator/metrics
. The metrics will have a name that has the format
ZUUL::EXCEPTION:errorCause:statusCode
.
For a general overview of how Zuul works, see the Zuul Wiki.
Zuul is implemented as a Servlet. For the general cases, Zuul is embedded into the Spring Dispatch mechanism. This lets Spring MVC be in control of the routing.
In this case, Zuul buffers requests.
If there is a need to go through Zuul without buffering requests (for example, for large file uploads), the Servlet is also installed outside of the Spring Dispatcher.
By default, the servlet has an address of /zuul
.
This path can be changed with the zuul.servlet-path
property.
To pass information between filters, Zuul uses a RequestContext
.
Its data is held in a ThreadLocal
specific to each request.
Information about where to route requests, errors, and the actual HttpServletRequest
and HttpServletResponse
are stored there.
The RequestContext
extends ConcurrentHashMap
, so anything can be stored in the context. FilterConstants
contains the keys used by the filters installed by Spring Cloud Netflix (more on these later).
Spring Cloud Netflix installs a number of filters, depending on which annotation was used to enable Zuul. @EnableZuulProxy
is a superset of @EnableZuulServer
. In other words, @EnableZuulProxy
contains all the filters installed by @EnableZuulServer
. The additional filters in the “proxy” enable routing functionality. If you want a “blank” Zuul, you should use @EnableZuulServer
.
@EnableZuulServer
creates a SimpleRouteLocator
that loads route definitions from Spring Boot configuration files.
The following filters are installed (as normal Spring Beans):
Pre filters:
ServletDetectionFilter
: Detects whether the request is through the Spring Dispatcher. Sets a boolean with a key of FilterConstants.IS_DISPATCHER_SERVLET_REQUEST_KEY
.FormBodyWrapperFilter
: Parses form data and re-encodes it for downstream requests.DebugFilter
: If the debug
request parameter is set, sets RequestContext.setDebugRouting()
and RequestContext.setDebugRequest()
to true
.
*Route filters:SendForwardFilter
: Forwards requests by using the Servlet RequestDispatcher
. The forwarding location is stored in the RequestContext
attribute, FilterConstants.FORWARD_TO_KEY
. This is useful for forwarding to endpoints in the current application.Post filters:
SendResponseFilter
: Writes responses from proxied requests to the current response.Error filters:
SendErrorFilter
: Forwards to /error
(by default) if RequestContext.getThrowable()
is not null. You can change the default forwarding path (/error
) by setting the error.path
property.Creates a DiscoveryClientRouteLocator
that loads route definitions from a DiscoveryClient
(such as Eureka) as well as from properties. A route is created for each serviceId
from the DiscoveryClient
. As new services are added, the routes are refreshed.
In addition to the filters described earlier, the following filters are installed (as normal Spring Beans):
Pre filters:
PreDecorationFilter
: Determines where and how to route, depending on the supplied RouteLocator
. It also sets various proxy-related headers for downstream requests.Route filters:
RibbonRoutingFilter
: Uses Ribbon, Hystrix, and pluggable HTTP clients to send requests. Service IDs are found in the RequestContext
attribute, FilterConstants.SERVICE_ID_KEY
. This filter can use different HTTP clients:
HttpClient
: The default client.OkHttpClient
v3: Enabled by having the com.squareup.okhttp3:okhttp
library on the classpath and setting ribbon.okhttp.enabled=true
.ribbon.restclient.enabled=true
. This client has limitations, including that it does not support the PATCH method, but it also has built-in retry.SimpleHostRoutingFilter
: Sends requests to predetermined URLs through an Apache HttpClient. URLs are found in RequestContext.getRouteHost()
.Most of the following "How to Write" examples below are included Sample Zuul Filters project. There are also examples of manipulating the request or response body in that repository.
This section includes the following examples:
Pre filters set up data in the RequestContext
for use in filters downstream.
The main use case is to set information required for route filters.
The following example shows a Zuul pre filter:
public class QueryParamPreFilter extends ZuulFilter { @Override public int filterOrder() { return PRE_DECORATION_FILTER_ORDER - 1; // run before PreDecoration } @Override public String filterType() { return PRE_TYPE; } @Override public boolean shouldFilter() { RequestContext ctx = RequestContext.getCurrentContext(); return !ctx.containsKey(FORWARD_TO_KEY) // a filter has already forwarded && !ctx.containsKey(SERVICE_ID_KEY); // a filter has already determined serviceId } @Override public Object run() { RequestContext ctx = RequestContext.getCurrentContext(); HttpServletRequest request = ctx.getRequest(); if (request.getParameter("sample") != null) { // put the serviceId in `RequestContext` ctx.put(SERVICE_ID_KEY, request.getParameter("foo")); } return null; } }
The preceding filter populates SERVICE_ID_KEY
from the sample
request parameter.
In practice, you should not do that kind of direct mapping. Instead, the service ID should be looked up from the value of sample
instead.
Now that SERVICE_ID_KEY
is populated, PreDecorationFilter
does not run and RibbonRoutingFilter
runs.
![]() | Tip |
---|---|
If you want to route to a full URL, call |
To modify the path to which routing filters forward, set the REQUEST_URI_KEY
.
Route filters run after pre filters and make requests to other services. Much of the work here is to translate request and response data to and from the model required by the client. The following example shows a Zuul route filter:
public class OkHttpRoutingFilter extends ZuulFilter { @Autowired private ProxyRequestHelper helper; @Override public String filterType() { return ROUTE_TYPE; } @Override public int filterOrder() { return SIMPLE_HOST_ROUTING_FILTER_ORDER - 1; } @Override public boolean shouldFilter() { return RequestContext.getCurrentContext().getRouteHost() != null && RequestContext.getCurrentContext().sendZuulResponse(); } @Override public Object run() { OkHttpClient httpClient = new OkHttpClient.Builder() // customize .build(); RequestContext context = RequestContext.getCurrentContext(); HttpServletRequest request = context.getRequest(); String method = request.getMethod(); String uri = this.helper.buildZuulRequestURI(request); Headers.Builder headers = new Headers.Builder(); Enumeration<String> headerNames = request.getHeaderNames(); while (headerNames.hasMoreElements()) { String name = headerNames.nextElement(); Enumeration<String> values = request.getHeaders(name); while (values.hasMoreElements()) { String value = values.nextElement(); headers.add(name, value); } } InputStream inputStream = request.getInputStream(); RequestBody requestBody = null; if (inputStream != null && HttpMethod.permitsRequestBody(method)) { MediaType mediaType = null; if (headers.get("Content-Type") != null) { mediaType = MediaType.parse(headers.get("Content-Type")); } requestBody = RequestBody.create(mediaType, StreamUtils.copyToByteArray(inputStream)); } Request.Builder builder = new Request.Builder() .headers(headers.build()) .url(uri) .method(method, requestBody); Response response = httpClient.newCall(builder.build()).execute(); LinkedMultiValueMap<String, String> responseHeaders = new LinkedMultiValueMap<>(); for (Map.Entry<String, List<String>> entry : response.headers().toMultimap().entrySet()) { responseHeaders.put(entry.getKey(), entry.getValue()); } this.helper.setResponse(response.code(), response.body().byteStream(), responseHeaders); context.setRouteHost(null); // prevent SimpleHostRoutingFilter from running return null; } }
The preceding filter translates Servlet request information into OkHttp3 request information, executes an HTTP request, and translates OkHttp3 response information to the Servlet response.
Post filters typically manipulate the response. The following filter adds a random UUID
as the X-Sample
header:
public class AddResponseHeaderFilter extends ZuulFilter { @Override public String filterType() { return POST_TYPE; } @Override public int filterOrder() { return SEND_RESPONSE_FILTER_ORDER - 1; } @Override public boolean shouldFilter() { return true; } @Override public Object run() { RequestContext context = RequestContext.getCurrentContext(); HttpServletResponse servletResponse = context.getResponse(); servletResponse.addHeader("X-Sample", UUID.randomUUID().toString()); return null; } }
![]() | Note |
---|---|
Other manipulations, such as transforming the response body, are much more complex and computationally intensive. |
If an exception is thrown during any portion of the Zuul filter lifecycle, the error filters are executed.
The SendErrorFilter
is only run if RequestContext.getThrowable()
is not null
.
It then sets specific javax.servlet.error.*
attributes in the request and forwards the request to the Spring Boot error page.
Zuul internally uses Ribbon for calling the remote URLs. By default, Ribbon clients are lazily loaded by Spring Cloud on first call. This behavior can be changed for Zuul by using the following configuration, which results eager loading of the child Ribbon related Application contexts at application startup time. The following example shows how to enable eager loading:
application.yml.
zuul: ribbon: eager-load: enabled: true
Do you have non-JVM languages with which you want to take advantage of Eureka, Ribbon, and Config Server? The Spring Cloud Netflix Sidecar was inspired by Netflix Prana. It includes an HTTP API to get all of the instances (by host and port) for a given service. You can also proxy service calls through an embedded Zuul proxy that gets its route entries from Eureka. The Spring Cloud Config Server can be accessed directly through host lookup or through the Zuul Proxy. The non-JVM application should implement a health check so the Sidecar can report to Eureka whether the app is up or down.
To include Sidecar in your project, use the dependency with a group ID of org.springframework.cloud
and artifact ID or spring-cloud-netflix-sidecar
.
To enable the Sidecar, create a Spring Boot application with @EnableSidecar
.
This annotation includes @EnableCircuitBreaker
, @EnableDiscoveryClient
, and @EnableZuulProxy
.
Run the resulting application on the same host as the non-JVM application.
To configure the side car, add sidecar.port
and sidecar.health-uri
to application.yml
.
The sidecar.port
property is the port on which the non-JVM application listens.
This is so the Sidecar can properly register the application with Eureka.
The sidecar.secure-port-enabled
options provides a way to enable secure port for traffic.
The sidecar.health-uri
is a URI accessible on the non-JVM application that mimics a Spring Boot health indicator.
It should return a JSON document that resembles the following:
health-uri-document.
{ "status":"UP" }
The following application.yml example shows sample configuration for a Sidecar application:
application.yml.
server: port: 5678 spring: application: name: sidecar sidecar: port: 8000 health-uri: http://localhost:8000/health.json
The API for the DiscoveryClient.getInstances()
method is /hosts/{serviceId}
.
The following example response for /hosts/customers
returns two instances on different hosts:
/hosts/customers.
[ { "host": "myhost", "port": 9000, "uri": "http://myhost:9000", "serviceId": "CUSTOMERS", "secure": false }, { "host": "myhost2", "port": 9000, "uri": "http://myhost2:9000", "serviceId": "CUSTOMERS", "secure": false } ]
This API is accessible to the non-JVM application (if the sidecar is on port 5678) at http://localhost:5678/hosts/{serviceId}
.
The Zuul proxy automatically adds routes for each service known in Eureka to /<serviceId>
, so the customers service is available at /customers
.
The non-JVM application can access the customer service at http://localhost:5678/customers
(assuming the sidecar is listening on port 5678).
If the Config Server is registered with Eureka, the non-JVM application can access it through the Zuul proxy.
If the serviceId
of the ConfigServer is configserver
and the Sidecar is on port 5678, then it can be accessed at http://localhost:5678/configserver.
Non-JVM applications can take advantage of the Config Server’s ability to return YAML documents. For example, a call to http://sidecar.local.spring.io:5678/configserver/default-master.yml might result in a YAML document resembling the following:
eureka: client: serviceUrl: defaultZone: http://localhost:8761/eureka/ password: password info: description: Spring Cloud Samples url: https://github.com/spring-cloud-samples
To enable the health check request to accept all certificates when using HTTPs set sidecar.accept-all-ssl-certificates
to `true.
Spring Cloud Netflix offers a variety of ways to make HTTP requests.
You can use a load balanced RestTemplate
, Ribbon, or Feign.
No matter how you choose to create your HTTP requests, there is always a chance that a request may fail.
When a request fails, you may want to have the request be retried automatically.
To do so when using Sping Cloud Netflix, you need to include Spring Retry on your application’s classpath.
When Spring Retry is present, load-balanced RestTemplates
, Feign, and Zuul automatically retry any failed requests (assuming your configuration allows doing so).
By default, no backoff policy is used when retrying requests.
If you would like to configure a backoff policy, you need to create a bean of type LoadBalancedRetryFactory
and override the createBackOffPolicy
method for a given service, as shown in the following example:
@Configuration public class MyConfiguration { @Bean LoadBalancedRetryFactory retryFactory() { return new LoadBalancedRetryFactory() { @Override public BackOffPolicy createBackOffPolicy(String service) { return new ExponentialBackOffPolicy(); } }; } }
When you use Ribbon with Spring Retry, you can control the retry functionality by configuring certain Ribbon properties.
To do so, set the client.ribbon.MaxAutoRetries
, client.ribbon.MaxAutoRetriesNextServer
, and client.ribbon.OkToRetryOnAllOperations
properties.
See the Ribbon documentation for a description of what these properties do.
![]() | Warning |
---|---|
Enabling |
In addition, you may want to retry requests when certain status codes are returned in the response.
You can list the response codes you would like the Ribbon client to retry by setting the clientName.ribbon.retryableStatusCodes
property, as shown in the following example:
clientName: ribbon: retryableStatusCodes: 404,502
You can also create a bean of type LoadBalancedRetryPolicy
and implement the retryableStatusCode
method to retry a request given the status code.
Spring Cloud Netflix automatically creates the HTTP client used by Ribbon, Feign, and Zuul for you.
However, you can also provide your own HTTP clients customized as you need them to be.
To do so, you can create a bean of type ClosableHttpClient
if you
are using the Apache Http Cient or OkHttpClient
if you are using OK HTTP.
![]() | Note |
---|---|
When you create your own HTTP client, you are also responsible for implementing the correct connection management strategies for these clients. Doing so improperly can result in resource management issues. |
Placing a module in maintenance mode means that the Spring Cloud team will no longer be adding new features to the module. We will fix blocker bugs and security issues, and we will also consider and review small pull requests from the community.
We intend to continue to support these modules for a period of at least a year from the general availability of the Greenwich release train.
The following Spring Cloud Netflix modules and corresponding starters will be placed into maintenance mode:
![]() | Note |
---|---|
This does not include the Eureka or concurrency-limits modules. |
1.0.0.BUILD-SNAPSHOT
This project provides OpenFeign integrations for Spring Boot apps through autoconfiguration and binding to the Spring Environment and other Spring programming model idioms.
Feign is a declarative web service client. It makes writing web service clients easier. To use Feign create an interface and annotate it. It has pluggable annotation support including Feign annotations and JAX-RS annotations. Feign also supports pluggable encoders and decoders. Spring Cloud adds support for Spring MVC annotations and for using the same HttpMessageConverters
used by default in Spring Web. Spring Cloud integrates Ribbon and Eureka to provide a load balanced http client when using Feign.
To include Feign in your project use the starter with group org.springframework.cloud
and artifact id spring-cloud-starter-openfeign
. See the Spring Cloud Project page
for details on setting up your build system with the current Spring Cloud Release Train.
Example spring boot app
@SpringBootApplication @EnableFeignClients public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
StoreClient.java.
@FeignClient("stores") public interface StoreClient { @RequestMapping(method = RequestMethod.GET, value = "/stores") List<Store> getStores(); @RequestMapping(method = RequestMethod.POST, value = "/stores/{storeId}", consumes = "application/json") Store update(@PathVariable("storeId") Long storeId, Store store); }
In the @FeignClient
annotation the String value ("stores" above) is
an arbitrary client name, which is used to create a Ribbon load
balancer (see below for details of Ribbon
support). You can also specify a URL using the url
attribute
(absolute value or just a hostname). The name of the bean in the
application context is the fully qualified name of the interface.
To specify your own alias value you can use the qualifier
value
of the @FeignClient
annotation.
The Ribbon client above will want to discover the physical addresses for the "stores" service. If your application is a Eureka client then it will resolve the service in the Eureka service registry. If you don’t want to use Eureka, you can simply configure a list of servers in your external configuration (see above for example).
A central concept in Spring Cloud’s Feign support is that of the named client. Each feign client is part of an ensemble of components that work together to contact a remote server on demand, and the ensemble has a name that you give it as an application developer using the @FeignClient
annotation. Spring Cloud creates a new ensemble as an
ApplicationContext
on demand for each named client using FeignClientsConfiguration
. This contains (amongst other things) an feign.Decoder
, a feign.Encoder
, and a feign.Contract
.
It is possible to override the name of that ensemble by using the contextId
attribute of the @FeignClient
annotation.
Spring Cloud lets you take full control of the feign client by declaring additional configuration (on top of the FeignClientsConfiguration
) using @FeignClient
. Example:
@FeignClient(name = "stores", configuration = FooConfiguration.class) public interface StoreClient { //.. }
In this case the client is composed from the components already in FeignClientsConfiguration
together with any in FooConfiguration
(where the latter will override the former).
![]() | Note |
---|---|
|
![]() | Note |
---|---|
The |
![]() | Note |
---|---|
Using |
![]() | Warning |
---|---|
Previously, using the |
Placeholders are supported in the name
and url
attributes.
@FeignClient(name = "${feign.name}", url = "${feign.url}") public interface StoreClient { //.. }
Spring Cloud Netflix provides the following beans by default for feign (BeanType
beanName: ClassName
):
Decoder
feignDecoder: ResponseEntityDecoder
(which wraps a SpringDecoder
)Encoder
feignEncoder: SpringEncoder
Logger
feignLogger: Slf4jLogger
Contract
feignContract: SpringMvcContract
Feign.Builder
feignBuilder: HystrixFeign.Builder
Client
feignClient: if Ribbon is enabled it is a LoadBalancerFeignClient
, otherwise the default feign client is used.The OkHttpClient and ApacheHttpClient feign clients can be used by setting feign.okhttp.enabled
or feign.httpclient.enabled
to true
, respectively, and having them on the classpath.
You can customize the HTTP client used by providing a bean of either ClosableHttpClient
when using Apache or OkHttpClient
when using OK HTTP.
Spring Cloud Netflix does not provide the following beans by default for feign, but still looks up beans of these types from the application context to create the feign client:
Logger.Level
Retryer
ErrorDecoder
Request.Options
Collection<RequestInterceptor>
SetterFactory
Creating a bean of one of those type and placing it in a @FeignClient
configuration (such as FooConfiguration
above) allows you to override each one of the beans described. Example:
@Configuration public class FooConfiguration { @Bean public Contract feignContract() { return new feign.Contract.Default(); } @Bean public BasicAuthRequestInterceptor basicAuthRequestInterceptor() { return new BasicAuthRequestInterceptor("user", "password"); } }
This replaces the SpringMvcContract
with feign.Contract.Default
and adds a RequestInterceptor
to the collection of RequestInterceptor
.
@FeignClient
also can be configured using configuration properties.
application.yml
feign: client: config: feignName: connectTimeout: 5000 readTimeout: 5000 loggerLevel: full errorDecoder: com.example.SimpleErrorDecoder retryer: com.example.SimpleRetryer requestInterceptors: - com.example.FooRequestInterceptor - com.example.BarRequestInterceptor decode404: false encoder: com.example.SimpleEncoder decoder: com.example.SimpleDecoder contract: com.example.SimpleContract
Default configurations can be specified in the @EnableFeignClients
attribute defaultConfiguration
in a similar manner as described above. The difference is that this configuration will apply to all feign clients.
If you prefer using configuration properties to configured all @FeignClient
, you can create configuration properties with default
feign name.
application.yml
feign: client: config: default: connectTimeout: 5000 readTimeout: 5000 loggerLevel: basic
If we create both @Configuration
bean and configuration properties, configuration properties will win.
It will override @Configuration
values. But if you want to change the priority to @Configuration
,
you can change feign.client.default-to-properties
to false
.
![]() | Note |
---|---|
If you need to use |
application.yml
# To disable Hystrix in Feign feign: hystrix: enabled: false # To set thread isolation to SEMAPHORE hystrix: command: default: execution: isolation: strategy: SEMAPHORE
If we want to create multiple feign clients with the same name or url
so that they would point to the same server but each with a different custom configuration then
we have to use contextId
attribute of the @FeignClient
in order to avoid name
collision of these configuration beans.
@FeignClient(contextId = "fooClient", name = "stores", configuration = FooConfiguration.class) public interface FooClient { //.. }
@FeignClient(contextId = "barClient", name = "stores", configuration = BarConfiguration.class) public interface BarClient { //.. }
In some cases it might be necessary to customize your Feign Clients in a way that is not possible using the methods above. In this case you can create Clients using the Feign Builder API. Below is an example which creates two Feign Clients with the same interface but configures each one with a separate request interceptor.
@Import(FeignClientsConfiguration.class) class FooController { private FooClient fooClient; private FooClient adminClient; @Autowired public FooController(Decoder decoder, Encoder encoder, Client client, Contract contract) { this.fooClient = Feign.builder().client(client) .encoder(encoder) .decoder(decoder) .contract(contract) .requestInterceptor(new BasicAuthRequestInterceptor("user", "user")) .target(FooClient.class, "http://PROD-SVC"); this.adminClient = Feign.builder().client(client) .encoder(encoder) .decoder(decoder) .contract(contract) .requestInterceptor(new BasicAuthRequestInterceptor("admin", "admin")) .target(FooClient.class, "http://PROD-SVC"); } }
![]() | Note |
---|---|
In the above example |
![]() | Note |
---|---|
|
![]() | Note |
---|---|
The Feign |
If Hystrix is on the classpath and feign.hystrix.enabled=true
, Feign will wrap all methods with a circuit breaker. Returning a com.netflix.hystrix.HystrixCommand
is also available. This lets you use reactive patterns (with a call to .toObservable()
or .observe()
or asynchronous use (with a call to .queue()
).
To disable Hystrix support on a per-client basis create a vanilla Feign.Builder
with the "prototype" scope, e.g.:
@Configuration public class FooConfiguration { @Bean @Scope("prototype") public Feign.Builder feignBuilder() { return Feign.builder(); } }
![]() | Warning |
---|---|
Prior to the Spring Cloud Dalston release, if Hystrix was on the classpath Feign would have wrapped all methods in a circuit breaker by default. This default behavior was changed in Spring Cloud Dalston in favor for an opt-in approach. |
Hystrix supports the notion of a fallback: a default code path that is executed when they circuit is open or there is an error. To enable fallbacks for a given @FeignClient
set the fallback
attribute to the class name that implements the fallback. You also need to declare your implementation as a Spring bean.
@FeignClient(name = "hello", fallback = HystrixClientFallback.class) protected interface HystrixClient { @RequestMapping(method = RequestMethod.GET, value = "/hello") Hello iFailSometimes(); } static class HystrixClientFallback implements HystrixClient { @Override public Hello iFailSometimes() { return new Hello("fallback"); } }
If one needs access to the cause that made the fallback trigger, one can use the fallbackFactory
attribute inside @FeignClient
.
@FeignClient(name = "hello", fallbackFactory = HystrixClientFallbackFactory.class) protected interface HystrixClient { @RequestMapping(method = RequestMethod.GET, value = "/hello") Hello iFailSometimes(); } @Component static class HystrixClientFallbackFactory implements FallbackFactory<HystrixClient> { @Override public HystrixClient create(Throwable cause) { return new HystrixClient() { @Override public Hello iFailSometimes() { return new Hello("fallback; reason was: " + cause.getMessage()); } }; } }
![]() | Warning |
---|---|
There is a limitation with the implementation of fallbacks in Feign and how Hystrix fallbacks work. Fallbacks are currently not supported for methods that return |
When using Feign with Hystrix fallbacks, there are multiple beans in the ApplicationContext
of the same type. This will cause @Autowired
to not work because there isn’t exactly one bean, or one marked as primary. To work around this, Spring Cloud Netflix marks all Feign instances as @Primary
, so Spring Framework will know which bean to inject. In some cases, this may not be desirable. To turn off this behavior set the primary
attribute of @FeignClient
to false.
@FeignClient(name = "hello", primary = false) public interface HelloClient { // methods here }
Feign supports boilerplate apis via single-inheritance interfaces. This allows grouping common operations into convenient base interfaces.
UserService.java.
public interface UserService { @RequestMapping(method = RequestMethod.GET, value ="/users/{id}") User getUser(@PathVariable("id") long id); }
UserResource.java.
@RestController public class UserResource implements UserService { }
UserClient.java.
package project.user; @FeignClient("users") public interface UserClient extends UserService { }
![]() | Note |
---|---|
It is generally not advisable to share an interface between a server and a client. It introduces tight coupling, and also actually doesn’t work with Spring MVC in its current form (method parameter mapping is not inherited). |
You may consider enabling the request or response GZIP compression for your Feign requests. You can do this by enabling one of the properties:
feign.compression.request.enabled=true feign.compression.response.enabled=true
Feign request compression gives you settings similar to what you may set for your web server:
feign.compression.request.enabled=true
feign.compression.request.mime-types=text/xml,application/xml,application/json
feign.compression.request.min-request-size=2048
These properties allow you to be selective about the compressed media types and minimum request threshold length.
A logger is created for each Feign client created. By default the name of the logger is the full class name of the interface used to create the Feign client. Feign logging only responds to the DEBUG
level.
application.yml.
logging.level.project.user.UserClient: DEBUG
The Logger.Level
object that you may configure per client, tells Feign how much to log. Choices are:
NONE
, No logging (DEFAULT).BASIC
, Log only the request method and URL and the response status code and execution time.HEADERS
, Log the basic information along with request and response headers.FULL
, Log the headers, body, and metadata for both requests and responses.For example, the following would set the Logger.Level
to FULL
:
@Configuration public class FooConfiguration { @Bean Logger.Level feignLoggerLevel() { return Logger.Level.FULL; } }
The OpenFeign @QueryMap
annotation provides support for POJOs to be used as
GET parameter maps. Unfortunately, the default OpenFeign QueryMap annotation is
incompatible with Spring because it lacks a value
property.
Spring Cloud OpenFeign provides an equivalent @SpringQueryMap
annotation, which
is used to annotate a POJO or Map parameter as a query parameter map.
For example, the Params
class defines parameters param1
and param2
:
// Params.java public class Params { private String param1; private String param2; // [Getters and setters omitted for brevity] }
The following feign client uses the Params
class by using the @SpringQueryMap
annotation:
@FeignClient("demo") public class DemoTemplate { @GetMapping(path = "/demo") String demoEndpoint(@SpringQueryMap Params params); }
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. * Port the business logic onto message brokers (such as RabbitMQ, Apache Kafka, Amazon Kinesis). * Interoperate between channel-based and non-channel-based application binding scenarios to support stateless and stateful computations by using Project Reactor’s Flux and Kafka Streams APIs. * Rely on the framework’s automatic content-type support for common use-cases. Extending to different data conversion types is possible.
You can try Spring Cloud Stream in less then 5 min even before you jump into any details by following this three-step guide.
We show you how to create a Spring Cloud Stream application that receives messages coming from the messaging middleware of your choice (more on this later) and logs received messages to the console.
We call it LoggingConsumer
.
While not very practical, it provides a good introduction to some of the main concepts
and abstractions, making it easier to digest the rest of this user guide.
The three steps are as follows:
To get started, visit the Spring Initializr. From there, you can generate our LoggingConsumer
application. To do so:
stream
.
When the “Cloud Stream” option should appears, select it.Select “Kafka” or “RabbitMQ”.
Basically, you choose the messaging middleware to which your application binds. We recommend using the one you have already installed or feel more comfortable with installing and running. Also, as you can see from the Initilaizer screen, there are a few other options you can choose. For example, you can choose Gradle as your build tool instead of Maven (the default).
In the Artifact field, type 'logging-consumer'.
The value of the Artifact field becomes the application name. If you chose RabbitMQ for the middleware, your Spring Initializr should now be as follows:
Click the Generate Project button.
Doing so downloads the zipped version of the generated project to your hard drive.
![]() | Tip |
---|---|
We encourage you to explore the many possibilities available in the Spring Initializr. It lets you create many different kinds of Spring applications. |
Now you can import the project into your IDE. Keep in mind that, depending on the IDE, you may need to follow a specific import procedure. For example, depending on how the project was generated (Maven or Gradle), you may need to follow specific import procedure (for example, in Eclipse or STS, you need to use File → Import → Maven → Existing Maven Project).
Once imported, the project must have no errors of any kind. Also, src/main/java
should contain com.example.loggingconsumer.LoggingConsumerApplication
.
Technically, at this point, you can run the application’s main class. It is already a valid Spring Boot application. However, it does not do anything, so we want to add some code.
Modify the com.example.loggingconsumer.LoggingConsumerApplication
class to look as follows:
@SpringBootApplication @EnableBinding(Sink.class) public class LoggingConsumerApplication { public static void main(String[] args) { SpringApplication.run(LoggingConsumerApplication.class, args); } @StreamListener(Sink.INPUT) public void handle(Person person) { System.out.println("Received: " + person); } public static class Person { private String name; public String getName() { return name; } public void setName(String name) { this.name = name; } public String toString() { return this.name; } } }
As you can see from the preceding listing:
Sink
binding (input-no-output) by using @EnableBinding(Sink.class)
.
Doing so signals to the framework to initiate binding to the messaging middleware, where it automatically creates the destination (that is, queue, topic, and others) that are bound to the Sink.INPUT
channel.handler
method to receive incoming messages of type Person
.
Doing so lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type Person
.You now have a fully functional Spring Cloud Stream application that does listens for messages.
From here, for simplicity, we assume you selected RabbitMQ in step one.
Assuming you have RabbitMQ installed and running, you can start the application by running its main
method in your IDE.
You should see following output:
--- [ main] c.s.b.r.p.RabbitExchangeQueueProvisioner : declaring queue for inbound: input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg, bound to: input --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:5672] --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#2a3a299:0/SimpleConnection@66c83fc8. . . . . . --- [ main] o.s.i.a.i.AmqpInboundChannelAdapter : started inbound.input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg . . . --- [ main] c.e.l.LoggingConsumerApplication : Started LoggingConsumerApplication in 2.531 seconds (JVM running for 2.897)
Go to the RabbitMQ management console or any other RabbitMQ client and send a message to input.anonymous.CbMIwdkJSBO1ZoPDOtHtCg
.
The anonymous.CbMIwdkJSBO1ZoPDOtHtCg
part represents the group name and is generated, so it is bound to be different in your environment.
For something more predictable, you can use an explicit group name by setting spring.cloud.stream.bindings.input.group=hello
(or whatever name you like).
The contents of the message should be a JSON representation of the Person
class, as follows:
{"name":"Sam Spade"}
Then, in your console, you should see:
Received: Sam Spade
You can also build and package your application into a boot jar (by using ./mvnw clean install
) and run the built JAR by using the java -jar
command.
Now you have a working (albeit very basic) Spring Cloud Stream application.
Spring Cloud Stream introduces a number of new features, enhancements, and changes. The following sections outline the most notable ones:
MeterRegistry
is also provided as a bean so that custom applications can autowire it to capture custom metrics.
See “Chapter 37, Metrics Emitter” for more details.RetryTemplate
, we now let you provide your own template, effectively overriding the one provided by the framework.
To use it, configure it as a @Bean
in your application.This version includes the following notable enhancements:
This change slims down the footprint of the deployed application in the event neither actuator nor web dependencies required. It also lets you switch between the reactive and conventional web paradigms by manually adding one of the following dependencies.
The following listing shows how to add the conventional web framework:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>
The following listing shows how to add the reactive web framework:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>
The following list shows how to add the actuator dependency:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency>
One of the core themes for verion 2.0 is improvements (in both consistency and performance) around content-type negotiation and message conversion. The following summary outlines the notable changes and improvements in this area. See the “Chapter 32, Content Type Negotiation” section for more details. Also this blog post contains more detail.
MessageConverter
objects.@StreamMessageConverter
annotation to provide custom MessageConverter
objects.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).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
.As of version 2.0, the following items have been deprecated:
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.
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)
.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.
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 @EnableBinding(Sink.class) public class VoteRecordingSinkApplication { public static void main(String[] args) { SpringApplication.run(VoteRecordingSinkApplication.class, args); } @StreamListener(Sink.INPUT) public void processVote(Vote vote) { votingService.recordVote(vote); } }
The @EnableBinding
annotation takes one or more interfaces as parameters (in this case, the parameter is a single Sink
interface).
An interface declares input and output channels.
Spring Cloud Stream provides the Source
, Sink
, and Processor
interfaces. You can also define your own interfaces.
The following listing shows the definition of the Sink
interface:
public interface Sink { String INPUT = "input"; @Input(Sink.INPUT) SubscribableChannel input(); }
The @Input
annotation identifies an input channel, through which received messages enter the application.
The @Output
annotation identifies an output channel, through which published messages leave the application.
The @Input
and @Output
annotations can take a channel name as a parameter.
If a name is not provided, the name of the annotated method is used.
Spring Cloud Stream creates an implementation of the interface for you. You can use this in the application by autowiring it, as shown in the following example (from a test case):
@RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(classes = VoteRecordingSinkApplication.class) @WebAppConfiguration @DirtiesContext public class StreamApplicationTests { @Autowired private Sink sink; @Test public void contextLoads() { assertNotNull(this.sink.input()); } }
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:
A Spring Cloud Stream application consists of a middleware-neutral core. The application communicates with the outside world through input and output channels injected into it by Spring Cloud Stream. Channels are connected to external brokers through middleware-specific Binder implementations.
Spring Cloud Stream 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.
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.
For example, deployers can dynamically choose, at runtime, the destinations (such as the Kafka topics or RabbitMQ exchanges) to which channels connect.
Such configuration can be provided through external configuration properties and in any form supported by Spring Boot (including application arguments, environment variables, and application.yml
or application.properties
files).
In the sink example from the Chapter 27, Introducing Spring Cloud Stream section, setting the spring.cloud.stream.bindings.input.destination
application property to raw-sensor-data
causes it to read from the raw-sensor-data
Kafka topic or from a queue bound to the raw-sensor-data
RabbitMQ exchange.
Spring Cloud Stream automatically detects and uses a binder found on the classpath. You can use different types of middleware with the same code. To do so, include a different binder at build time. For more complex use cases, you can also package multiple binders with your application and have it choose the binder( and even whether to use different binders for different channels) at runtime.
Communication between applications follows a publish-subscribe model, where data is broadcast through shared topics. This can be seen in the following figure, which shows a typical deployment for a set of interacting Spring Cloud Stream applications.
Data reported by sensors to an HTTP endpoint is sent to a common destination named raw-sensor-data
.
From the destination, it is independently processed by a microservice application that computes time-windowed averages and by another microservice application that ingests the raw data into HDFS (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. You can then add another application that interprets the same flow of averages for fault detection. Doing all communication through shared topics rather than point-to-point queues reduces coupling between microservices.
While the concept of publish-subscribe messaging is not new, Spring Cloud Stream takes the extra step of making it an opinionated choice for its application model. By using native middleware support, Spring Cloud Stream also simplifies use of the publish-subscribe model across different platforms.
While the publish-subscribe model makes it easy to connect applications through shared topics, the ability to scale up by creating multiple instances of a given application is equally important. When doing 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
.
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.
Two types of consumer are supported:
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.
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 |
---|---|
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. When scaling up a Spring Cloud Stream application, you must specify a consumer group for each of its input bindings. Doing so prevents the application’s instances from receiving duplicate messages (unless that behavior is desired, which is unusual).
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).
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 |
---|---|
To set up a partitioned processing scenario, you must configure both the data-producing and the data-consuming ends. |
To understand the programming model, you should be familiar with the following core concepts:
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 some help in the form of minimalistic yet required set of instructions from the user, which typically come in the form of some type of configuration.
While it is out of scope of this section to discuss all of the available binder and binding configuration options (the rest of the manual covers them extensively), Destination Binding does require special attention. The next section discusses it in detail.
As stated earlier, Destination Bindings provide a bridge between the external messaging system and application-provided Producers and Consumers.
Applying the @EnableBinding annotation to one of the application’s configuration classes defines a destination binding.
The @EnableBinding
annotation itself is meta-annotated with @Configuration
and triggers the configuration of the Spring Cloud Stream infrastructure.
The following example shows a fully configured and functioning Spring Cloud Stream application that receives the payload of the message from the INPUT
destination as a String
type (see Chapter 32, Content Type Negotiation section), logs it to the console and sends it to the OUTPUT
destination after converting it to upper case.
@SpringBootApplication @EnableBinding(Processor.class) public class MyApplication { public static void main(String[] args) { SpringApplication.run(MyApplication.class, args); } @StreamListener(Processor.INPUT) @SendTo(Processor.OUTPUT) public String handle(String value) { System.out.println("Received: " + value); return value.toUpperCase(); } }
As you can see the @EnableBinding
annotation can take one or more interface classes as parameters. The parameters are referred to as bindings,
and they contain methods representing bindable components.
These components are typically message channels (see Spring Messaging)
for channel-based binders (such as Rabbit, Kafka, and others). However other types of bindings can
provide support for the native features of the corresponding technology. For example Kafka Streams binder (formerly known as KStream) allows native bindings directly to Kafka Streams
(see Kafka Streams for more details).
Spring Cloud Stream already provides binding interfaces for typical message exchange contracts, which include:
public interface Sink { String INPUT = "input"; @Input(Sink.INPUT) SubscribableChannel input(); }
public interface Source { String OUTPUT = "output"; @Output(Source.OUTPUT) MessageChannel output(); }
public interface Processor extends Source, Sink {}
While the preceding example satisfies the majority of cases, you can also define your own contracts by defining your own bindings interfaces and use @Input
and @Output
annotations to identify the actual bindable components.
For example:
public interface Barista { @Input SubscribableChannel orders(); @Output MessageChannel hotDrinks(); @Output MessageChannel coldDrinks(); }
Using the interface shown in the preceding example as a parameter to @EnableBinding
triggers the creation of the three bound channels named orders
, hotDrinks
, and coldDrinks
,
respectively.
You can provide as many binding interfaces as you need, as arguments to the @EnableBinding
annotation, as shown in the following example:
@EnableBinding(value = { Orders.class, Payment.class })
In Spring Cloud Stream, the bindable MessageChannel
components are the Spring Messaging MessageChannel
(for outbound) and its extension, SubscribableChannel
,
(for inbound).
Pollable Destination Binding
While the previously described bindings support event-based message consumption, sometimes you need more control, such as rate of consumption.
Starting with version 2.0, you can now bind a pollable consumer:
The following example shows how to bind a pollable consumer:
public interface PolledBarista { @Input PollableMessageSource orders(); . . . }
In this case, an implementation of PollableMessageSource
is bound to the orders
“channel”. See Section 29.3.5, “Using Polled Consumers” for more details.
Customizing Channel Names
By using the @Input
and @Output
annotations, you can specify a customized channel name for the channel, as shown in the following example:
public interface Barista { @Input("inboundOrders") SubscribableChannel orders(); }
In the preceding example, the created bound channel is named inboundOrders
.
Normally, you need not access individual channels or bindings directly (other then configuring them via @EnableBinding
annotation). However there may be
times, such as testing or other corner cases, when you do.
Aside from generating channels for each binding and registering them as Spring beans, for each bound interface, Spring Cloud Stream generates a bean that implements the interface. That means you can have access to the interfaces representing the bindings or individual channels by auto-wiring either in your application, as shown in the following two examples:
Autowire Binding interface
@Autowire private Source source public void sayHello(String name) { source.output().send(MessageBuilder.withPayload(name).build()); }
Autowire individual channel
@Autowire private MessageChannel output; public void sayHello(String name) { output.send(MessageBuilder.withPayload(name).build()); }
You can also use standard Spring’s @Qualifier
annotation for cases when channel names are customized or in multiple-channel scenarios that require specifically named channels.
The following example shows how to use the @Qualifier annotation in this way:
@Autowire @Qualifier("myChannel") private MessageChannel output;
You can write a Spring Cloud Stream application by using either Spring Integration annotations or Spring Cloud Stream native annotation.
Spring Cloud Stream is built on the concepts and patterns defined by Enterprise Integration Patterns and relies in its internal implementation on an already established and popular implementation of Enterprise Integration Patterns within the Spring portfolio of projects: Spring Integration framework.
So its only natural for it to support the foundation, semantics, and configuration options that are already established by Spring Integration
For example, you can attach the output channel of a Source
to a MessageSource
and use the familiar @InboundChannelAdapter
annotation, as follows:
@EnableBinding(Source.class) public class TimerSource { @Bean @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "10", maxMessagesPerPoll = "1")) public MessageSource<String> timerMessageSource() { return () -> new GenericMessage<>("Hello Spring Cloud Stream"); } }
Similarly, you can use @Transformer or @ServiceActivator while providing an implementation of a message handler method for a Processor binding contract, as shown in the following example:
@EnableBinding(Processor.class) public class TransformProcessor { @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public Object transform(String message) { return message.toUpperCase(); } }
![]() | Note |
---|---|
While this may be skipping ahead a bit, it is important to understand that, when you consume from the same binding using |
Complementary to its Spring Integration support, Spring Cloud Stream provides its own @StreamListener
annotation, modeled after other Spring Messaging annotations
(@MessageMapping
, @JmsListener
, @RabbitListener
, and others) and provides conviniences, such as content-based routing and others.
@EnableBinding(Sink.class) public class VoteHandler { @Autowired VotingService votingService; @StreamListener(Sink.INPUT) public void handle(Vote vote) { votingService.record(vote); } }
As with other Spring Messaging methods, method arguments can be annotated with @Payload
, @Headers
, and @Header
.
For methods that return data, you must use the @SendTo
annotation to specify the output binding destination for data returned by the method, as shown in the following example:
@EnableBinding(Processor.class) public class TransformProcessor { @Autowired VotingService votingService; @StreamListener(Processor.INPUT) @SendTo(Processor.OUTPUT) public VoteResult handle(Vote vote) { return votingService.record(vote); } }
Spring Cloud Stream supports dispatching messages to multiple handler methods annotated with @StreamListener
based on conditions.
In order to be eligible to support conditional dispatching, a method must satisfy the follow conditions:
The condition is specified by a SpEL expression in the condition
argument of the annotation and is evaluated for each message.
All the handlers that match the condition are invoked in the same thread, and no assumption must be made about the order in which the invocations take place.
In the following example of a @StreamListener
with dispatching conditions, all the messages bearing a header type
with the value bogey
are dispatched to the
receiveBogey
method, and all the messages bearing a header type
with the value bacall
are dispatched to the receiveBacall
method.
@EnableBinding(Sink.class) @EnableAutoConfiguration public static class TestPojoWithAnnotatedArguments { @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'") public void receiveBogey(@Payload BogeyPojo bogeyPojo) { // handle the message } @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bacall'") public void receiveBacall(@Payload BacallPojo bacallPojo) { // handle the message } }
Content Type Negotiation in the Context of condition
It is important to understand some of the mechanics behind content-based routing using the condition
argument of @StreamListener
, especially in the context of the type of the message as a whole.
It may also help if you familiarize yourself with the Chapter 32, Content Type Negotiation before you proceed.
Consider the following scenario:
@EnableBinding(Sink.class) @EnableAutoConfiguration public static class CatsAndDogs { @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Dog'") public void bark(Dog dog) { // handle the message } @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Cat'") public void purr(Cat cat) { // handle the message } }
The preceding code is perfectly valid. It compiles and deploys without any issues, yet it never produces the result you expect.
That is because you are testing something that does not yet exist in a state you expect. That is because the payload of the message is not yet converted from the
wire format (byte[]
) to the desired type.
In other words, it has not yet gone through the type conversion process described in the Chapter 32, Content Type Negotiation.
So, unless you use a SPeL expression that evaluates raw data (for example, the value of the first byte in the byte array), use message header-based expressions
(such as condition = "headers['type']=='dog'"
).
![]() | Note |
---|---|
At the moment, dispatching through |
Since Spring Cloud Stream v2.1, another alternative for defining stream handlers and sources is to use build-in
support for Spring Cloud Function where they can be expressed as beans of
type java.util.function.[Supplier/Function/Consumer]
.
To specify which functional bean to bind to the external destination(s) exposed by the bindings, you must provide spring.cloud.stream.function.definition
property.
Here is the example of the Processor application exposing message handler as java.util.function.Function
@SpringBootApplication @EnableBinding(Processor.class) public class MyFunctionBootApp { public static void main(String[] args) { SpringApplication.run(MyFunctionBootApp.class, "--spring.cloud.stream.function.definition=toUpperCase"); } @Bean public Function<String, String> toUpperCase() { return s -> s.toUpperCase(); } }
In the above you we simply define a bean of type java.util.function.Function
called toUpperCase and identify it as a bean to be used as message handler
whose 'input' and 'output' must be bound to the external destinations exposed by the Processor binding.
Below are the examples of simple functional applications to support Source, Processor and Sink.
Here is the example of a Source application defined as java.util.function.Supplier
@SpringBootApplication @EnableBinding(Source.class) public static class SourceFromSupplier { public static void main(String[] args) { SpringApplication.run(SourceFromSupplier.class, "--spring.cloud.stream.function.definition=date"); } @Bean public Supplier<Date> date() { return () -> new Date(12345L); } }
Here is the example of a Processor application defined as java.util.function.Function
@SpringBootApplication @EnableBinding(Processor.class) public static class ProcessorFromFunction { public static void main(String[] args) { SpringApplication.run(ProcessorFromFunction.class, "--spring.cloud.stream.function.definition=toUpperCase"); } @Bean public Function<String, String> toUpperCase() { return s -> s.toUpperCase(); } }
Here is the example of a Sink application defined as java.util.function.Consumer
@EnableAutoConfiguration @EnableBinding(Sink.class) public static class SinkFromConsumer { public static void main(String[] args) { SpringApplication.run(SinkFromConsumer.class, "--spring.cloud.stream.function.definition=sink"); } @Bean public Consumer<String> sink() { return System.out::println; } }
Using this programming model you can also benefit from functional composition where you can dynamically compose complex handlers from a set of simple functions. As an example let’s add the following function bean to the application defined above
@Bean public Function<String, String> wrapInQuotes() { return s -> "\"" + s + "\""; }
and modify the spring.cloud.stream.function.definition
property to reflect your intention to compose a new function from both ‘toUpperCase’ and ‘wrapInQuotes’.
To do that Spring Cloud Function allows you to use |
(pipe) symbol. So to finish our example our property will now look like this:
—spring.cloud.stream.function.definition=toUpperCase|wrapInQuotes
When using polled consumers, you poll the PollableMessageSource
on demand.
Consider the following example of a polled consumer:
public interface PolledConsumer { @Input PollableMessageSource destIn(); @Output MessageChannel destOut(); }
Given the polled consumer in the preceding example, you might use it as follows:
@Bean public ApplicationRunner poller(PollableMessageSource destIn, MessageChannel destOut) { return args -> { while (someCondition()) { try { if (!destIn.poll(m -> { String newPayload = ((String) m.getPayload()).toUpperCase(); destOut.send(new GenericMessage<>(newPayload)); })) { Thread.sleep(1000); } } catch (Exception e) { // handle failure } } }; }
The PollableMessageSource.poll()
method takes a MessageHandler
argument (often a lambda expression, as shown here).
It returns true
if the message was received and successfully processed.
As with message-driven consumers, if the MessageHandler
throws an exception, messages are published to error channels, as discussed in “???”.
Normally, the poll()
method acknowledges the message when the MessageHandler
exits.
If the method exits abnormally, the message is rejected (not re-queued), but see the section called “Handling Errors”.
You can override that behavior by taking responsibility for the acknowledgment, as shown in the following example:
@Bean public ApplicationRunner poller(PollableMessageSource dest1In, MessageChannel dest2Out) { return args -> { while (someCondition()) { if (!dest1In.poll(m -> { StaticMessageHeaderAccessor.getAcknowledgmentCallback(m).noAutoAck(); // e.g. hand off to another thread which can perform the ack // or acknowledge(Status.REQUEUE) })) { Thread.sleep(1000); } } }; }
![]() | Important |
---|---|
You must |
![]() | Important |
---|---|
Some messaging systems (such as Apache Kafka) maintain a simple offset in a log. If a delivery fails and is re-queued with |
There is also an overloaded poll
method, for which the definition is as follows:
poll(MessageHandler handler, ParameterizedTypeReference<?> type)
The type
is a conversion hint that allows the incoming message payload to be converted, as shown in the following example:
boolean result = pollableSource.poll(received -> { Map<String, Foo> payload = (Map<String, Foo>) received.getPayload(); ... }, new ParameterizedTypeReference<Map<String, Foo>>() {});
By default, an error channel is configured for the pollable source; if the callback throws an exception, an ErrorMessage
is sent to the error channel (<destination>.<group>.errors
); this error channel is also bridged to the global Spring Integration errorChannel
.
You can subscribe to either error channel with a @ServiceActivator
to handle errors; without a subscription, the error will simply be logged and the message will be acknowledged as successful.
If the error channel service activator throws an exception, the message will be rejected (by default) and won’t be redelivered.
If the service activator throws a RequeueCurrentMessageException
, the message will be requeued at the broker and will be again retrieved on a subsequent poll.
If the listener throws a RequeueCurrentMessageException
directly, the message will be requeued, as discussed above, and will not be sent to the error channels.
Errors happen, and Spring Cloud Stream provides several flexible mechanisms to handle them. The error handling comes in two flavors:
Spring Cloud Stream uses the Spring Retry library to facilitate successful message processing. See Section 29.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).
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.
For each input binding, Spring Cloud Stream creates a dedicated error channel with the following semantics <destinationName>.errors
.
![]() | Note |
---|---|
The |
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!"); } @ServiceActivator(inputChannel = Processor.INPUT + ".myGroup.errors") //channel name 'input.myGroup.errors' public void error(Message<?> message) { System.out.println("Handling ERROR: " + message); }
In the preceding example the destination name is input.myGroup
and the dedicated error channel name is input.myGroup.errors
.
![]() | Note |
---|---|
The use of @StreamListener annotation is intended specifically to define bindings that bridge internal channels and external destinations. Given that the destination specific error channel does NOT have an associated external destination, such channel is a prerogative of Spring Integration (SI). This means that the handler for such destination must be defined using one of the SI handler annotations (i.e., @ServiceActivator, @Transformer etc.). |
![]() | Note |
---|---|
If |
Also, in the event you are binding to the existing destination such as:
spring.cloud.stream.bindings.input.destination=myFooDestination spring.cloud.stream.bindings.input.group=myGroup
the full destination name is myFooDestination.myGroup
and then the dedicated error channel name is myFooDestination.myGroup.errors
.
Back to the example…
The handle(..)
method, which subscribes to the channel named input
, throws an exception. Given there is also a subscriber to the error channel input.myGroup.errors
all error messages are handled by this subscriber.
If you have multiple bindings, you may want to have a single error handler. Spring Cloud Stream automatically provides support for
a global error channel by bridging each individual error channel to the channel named errorChannel
, allowing a single subscriber to handle all errors,
as shown in the following example:
@StreamListener("errorChannel") public void error(Message<?> message) { System.out.println("Handling ERROR: " + message); }
This may be a convenient option if error handling logic is the same regardless of which handler produced the error.
System-level error handling implies that the errors are communicated back to the messaging system and, given that not every messaging system is the same, the capabilities may differ from binder to binder.
That said, in this section we explain the general idea behind system level error handling and use Rabbit binder as an example. NOTE: Kafka binder provides similar support, although some configuration properties do differ. Also, for more details and configuration options, see the individual binder’s documentation.
If no internal error handlers are configured, the errors propagate to the binders, and the binders subsequently propagate those errors back to the messaging system. Depending on the capabilities of the messaging system such a system may drop the message, re-queue the message for re-processing or send the failed message to DLQ. Both Rabbit and Kafka support these concepts. However, other binders may not, so refer to your individual binder’s documentation for details on supported system-level error-handling options.
By default, if no additional system-level configuration is provided, the messaging system drops the failed message. While acceptable in some cases, for most cases, it is not, and we need some recovery mechanism to avoid message loss.
DLQ allows failed messages to be sent to a special destination: - Dead Letter Queue.
When configured, failed messages are sent to this destination for subsequent re-processing or auditing and reconciliation.
For example, continuing on the previous example and to set up the DLQ with Rabbit binder, you need to set the following property:
spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true
Keep in mind that, in the above property, input
corresponds to the name of the input destination binding.
The consumer
indicates that it is a consumer property and auto-bind-dlq
instructs the binder to configure DLQ for input
destination, which results in an additional Rabbit queue named input.myGroup.dlq
.
Once configured, all failed messages are routed to this queue with an error message similar to the following:
delivery_mode: 1 headers: x-death: count: 1 reason: rejected queue: input.hello time: 1522328151 exchange: routing-keys: input.myGroup Payload {"name”:"Bob"}
As you can see from the above, your original message is preserved for further actions.
However, one thing you may have noticed is that there is limited information on the original issue with the message processing. For example, you do not see a stack trace corresponding to the original error. To get more relevant information about the original error, you must set an additional property:
spring.cloud.stream.rabbit.bindings.input.consumer.republish-to-dlq=true
Doing so forces the internal error handler to intercept the error message and add additional information to it before publishing it to DLQ. Once configured, you can see that the error message contains more information relevant to the original error, as follows:
delivery_mode: 2 headers: x-original-exchange: x-exception-message: has an error x-original-routingKey: input.myGroup x-exception-stacktrace: org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.messaging.MessagingException: has an error, failedMessage=GenericMessage [payload=byte[15], headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=input.hello, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=input.hello, amqp_redelivered=false, id=a15231e6-3f80-677b-5ad7-d4b1e61e486e, amqp_consumerTag=amq.ctag-skBFapilvtZhDsn0k3ZmQg, contentType=application/json, timestamp=1522327846136}] at org.spring...integ...han...MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:107) at. . . . . Payload {"name”:"Bob"}
This effectively combines application-level and system-level error handling to further assist with downstream troubleshooting mechanics.
As mentioned earlier, the currently supported binders (Rabbit and Kafka) rely on RetryTemplate
to facilitate successful message processing. See Section 29.4.3, “Retry Template” for details.
However, for cases when max-attempts
property is set to 1, internal reprocessing of the message is disabled. At this point, you can facilitate message re-processing (re-tries)
by instructing the messaging system to re-queue the failed message. Once re-queued, the failed message is sent back to the original handler, essentially creating a retry loop.
This option may be feasible for cases where the nature of the error is related to some sporadic yet short-term unavailability of some resource.
To accomplish that, you must set the following properties:
spring.cloud.stream.bindings.input.consumer.max-attempts=1 spring.cloud.stream.rabbit.bindings.input.consumer.requeue-rejected=true
In the preceding example, the max-attempts
set to 1 essentially disabling internal re-tries and requeue-rejected
(short for requeue rejected messages) is set to true
.
Once set, the failed message is resubmitted to the same handler and loops continuously or until the handler throws AmqpRejectAndDontRequeueException
essentially allowing you to build your own re-try logic within the handler itself.
The RetryTemplate
is part of the Spring Retry library.
While it is out of scope of this document to cover all of the capabilities of the RetryTemplate
, we will mention the following consumer properties that are specifically related to
the RetryTemplate
:
The number of attempts to process the message.
Default: 3.
The backoff initial interval on retry.
Default 1000 milliseconds.
The maximum backoff interval.
Default 10000 milliseconds.
The backoff multiplier.
Default 2.0.
Whether exceptions thrown by the listener that are not listed in the retryableExceptions
are retryable.
Default: true
.
A map of Throwable class names in the key and a boolean in the value.
Specify those exceptions (and subclasses) that will or won’t be retried.
Also see defaultRetriable
.
Example: spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false
.
Default: empty.
While the preceding settings are sufficient for majority of the customization requirements, they may not satisfy certain complex requirements at, which
point you may want to provide your own instance of the RetryTemplate
. To do so configure it as a bean in your application configuration. The application provided
instance will override the one provided by the framework. Also, to avoid conflicts you must qualify the instance of the RetryTemplate
you want to be used by the binder
as @StreamRetryTemplate
. For example,
@StreamRetryTemplate public RetryTemplate myRetryTemplate() { return new RetryTemplate(); }
As you can see from the above example you don’t need to annotate it with @Bean
since @StreamRetryTemplate
is a qualified @Bean
.
Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows.
Support for reactive APIs is available through spring-cloud-stream-reactive
, which needs to be added explicitly to your project.
The programming model with reactive APIs is declarative. Instead of specifying how each individual message should be handled, you can use operators that describe functional transformations from inbound to outbound data flows.
At present Spring Cloud Stream supports the only the Reactor API. In the future, we intend to support a more generic model based on Reactive Streams.
The reactive programming model also uses the @StreamListener
annotation for setting up reactive handlers.
The differences are that:
@StreamListener
annotation must not specify an input or output, as they are provided as arguments and return values from the method.@Input
and @Output
, indicating which input or output the incoming and outgoing data flows connect to, respectively.@Output
, indicating the input where data should be sent.![]() | Note |
---|---|
Reactive programming support requires Java 1.8. |
![]() | Note |
---|---|
As of Spring Cloud Stream 1.1.1 and later (starting with release train Brooklyn.SR2), reactive programming support requires the use of Reactor 3.0.4.RELEASE and higher.
Earlier Reactor versions (including 3.0.1.RELEASE, 3.0.2.RELEASE and 3.0.3.RELEASE) are not supported.
|
![]() | Note |
---|---|
The use of term, “reactive”, currently refers to the reactive APIs being used and not to the execution model being reactive (that is, the bound endpoints still use a 'push' rather than a 'pull' model). While some backpressure support is provided by the use of Reactor, we do intend, in a future release, to support entirely reactive pipelines by the use of native reactive clients for the connected middleware. |
A Reactor-based handler can have the following argument types:
@Input
, it supports the Reactor Flux
type.
The parameterization of the inbound Flux follows the same rules as in the case of individual message handling: It can be the entire Message
, a POJO that can be the Message
payload, or a POJO that is the result of a transformation based on the Message
content-type header. Multiple inputs are provided.Output
, it supports the FluxSender
type, which connects a Flux
produced by the method with an output. Generally speaking, specifying outputs as arguments is only recommended when the method can have multiple outputs.A Reactor-based handler supports a return type of Flux
. In that case, it must be annotated with @Output
. We recommend using the return value of the method when a single output Flux
is available.
The following example shows a Reactor-based Processor
:
@EnableBinding(Processor.class) @EnableAutoConfiguration public static class UppercaseTransformer { @StreamListener @Output(Processor.OUTPUT) public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) { return input.map(s -> s.toUpperCase()); } }
The same processor using output arguments looks like the following example:
@EnableBinding(Processor.class) @EnableAutoConfiguration public static class UppercaseTransformer { @StreamListener public void receive(@Input(Processor.INPUT) Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) { output.send(input.map(s -> s.toUpperCase())); } }
Spring Cloud Stream reactive support also provides the ability for creating reactive sources through the @StreamEmitter
annotation.
By using the @StreamEmitter
annotation, a regular source may be converted to a reactive one.
@StreamEmitter
is a method level annotation that marks a method to be an emitter to outputs declared with @EnableBinding
.
You cannot use the @Input
annotation along with @StreamEmitter
, as the methods marked with this annotation are not listening for any input. Rather, methods marked with @StreamEmitter
generate output.
Following the same programming model used in @StreamListener
, @StreamEmitter
also allows flexible ways of using the @Output
annotation, depending on whether the method has any arguments, a return type, and other considerations.
The remainder of this section contains examples of using the @StreamEmitter
annotation in various styles.
The following example emits the Hello, World
message every millisecond and publishes to a Reactor Flux
:
@EnableBinding(Source.class) @EnableAutoConfiguration public static class HelloWorldEmitter { @StreamEmitter @Output(Source.OUTPUT) public Flux<String> emit() { return Flux.intervalMillis(1) .map(l -> "Hello World"); } }
In the preceding example, the resulting messages in the Flux
are sent to the output channel of the Source
.
The next example is another flavor of an @StreamEmmitter
that sends a Reactor Flux
.
Instead of returning a Flux
, the following method uses a FluxSender
to programmatically send a Flux
from a source:
@EnableBinding(Source.class) @EnableAutoConfiguration public static class HelloWorldEmitter { @StreamEmitter @Output(Source.OUTPUT) public void emit(FluxSender output) { output.send(Flux.intervalMillis(1) .map(l -> "Hello World")); } }
The next example is exactly same as the above snippet in functionality and style.
However, instead of using an explicit @Output
annotation on the method, it uses the annotation on the method parameter.
@EnableBinding(Source.class) @EnableAutoConfiguration public static class HelloWorldEmitter { @StreamEmitter public void emit(@Output(Source.OUTPUT) FluxSender output) { output.send(Flux.intervalMillis(1) .map(l -> "Hello World")); } }
The last example in this section is yet another flavor of writing reacting sources by using the Reactive Streams Publisher API and taking advantage of the support for it in Spring Integration Java DSL.
The Publisher
in the following example still uses Reactor Flux
under the hood, but, from an application perspective, that is transparent to the user and only needs Reactive Streams and Java DSL for Spring Integration:
@EnableBinding(Source.class) @EnableAutoConfiguration public static class HelloWorldEmitter { @StreamEmitter @Output(Source.OUTPUT) @Bean public Publisher<Message<String>> emit() { return IntegrationFlows.from(() -> new GenericMessage<>("Hello World"), e -> e.poller(p -> p.fixedDelay(1))) .toReactivePublisher(); } }
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.
The following image shows the general relationship of producers and 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.
When invoking the bindConsumer()
method, the first parameter is the destination name, and a second parameter provides the name of a logical group of consumers.
Each group that is represented by consumer bindings for a given destination receives a copy of each message that a producer sends to that destination (that is, it follows normal publish-subscribe semantics).
If there are multiple consumer instances bound with the same group name, then messages are load-balanced across those consumer instances so that each message sent by a producer is consumed by only a single consumer instance within each group (that is, it follows normal queueing semantics).
The Binder SPI consists of a number of interfaces, out-of-the box utility classes, and discovery strategies that provide a pluggable mechanism for connecting to external middleware.
The key point of the SPI is the Binder
interface, which is a strategy for connecting inputs and outputs to external middleware. The following listing shows the definnition of the Binder
interface:
public interface Binder<T, C extends ConsumerProperties, P extends ProducerProperties> { Binding<T> bindConsumer(String name, String group, T inboundBindTarget, C consumerProperties); Binding<T> bindProducer(String name, T outboundBindTarget, P producerProperties); }
The interface is parameterized, offering a number of extension points:
MessageChannel
is supported, but this is intended to be used as an extension point in the future.A typical binder implementation consists of the following:
Binder
interface;@Configuration
class that creates a bean of type Binder
along with the middleware connection infrastructure.A META-INF/spring.binders
file found on the classpath containing one or more binder definitions, as shown in the following example:
kafka:\ org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration
Spring Cloud Stream relies on implementations of the Binder SPI to perform the task of connecting channels to message brokers. Each Binder implementation typically connects to one type of messaging system.
By default, Spring Cloud Stream relies on Spring Boot’s auto-configuration to configure the binding process. If a single Binder implementation is found on the classpath, Spring Cloud Stream automatically uses it. For example, a Spring Cloud Stream project that aims to bind only to RabbitMQ can add the following dependency:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>
For the specific Maven coordinates of other binder dependencies, see the documentation of that binder implementation.
When multiple binders are present on the classpath, the application must indicate which binder is to be used for each channel binding.
Each binder configuration contains a META-INF/spring.binders
file, which is a simple properties file, as shown in the following example:
rabbit:\ org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration
Similar files exist for the other provided binder implementations (such as Kafka), and custom binder implementations are expected to provide them as well.
The key represents an identifying name for the binder implementation, whereas the value is a comma-separated list of configuration classes that each contain one and only one bean definition of type org.springframework.cloud.stream.binder.Binder
.
Binder selection can either be performed globally, using the spring.cloud.stream.defaultBinder
property (for example, spring.cloud.stream.defaultBinder=rabbit
) or individually, by configuring the binder on each channel binding.
For instance, a processor application (that has channels named input
and output
for read and write respectively) that reads from Kafka and writes to RabbitMQ can specify the following configuration:
spring.cloud.stream.bindings.input.binder=kafka spring.cloud.stream.bindings.output.binder=rabbit
By default, binders share the application’s Spring Boot auto-configuration, so that one instance of each binder found on the classpath is created. If your application should connect to more than one broker of the same type, you can specify multiple binder configurations, each with different environment settings.
![]() | Note |
---|---|
Turning on explicit binder configuration disables the default binder configuration process altogether.
If you do so, all binders in use must be included in the configuration.
Frameworks that intend to use Spring Cloud Stream transparently may create binder configurations that can be referenced by name, but they do not affect the default binder configuration.
In order to do so, a binder configuration may have its |
The following example shows a typical configuration for a processor application that connects to two RabbitMQ broker instances:
spring: cloud: stream: bindings: input: destination: thing1 binder: rabbit1 output: destination: thing2 binder: rabbit2 binders: rabbit1: type: rabbit environment: spring: rabbitmq: host: <host1> rabbit2: type: rabbit environment: spring: rabbitmq: host: <host2>
Since version 2.0, Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints.
Starting with version 2.0 actuator and web are optional, you must first add one of the web dependencies as well as add the actuator dependency manually. The following example shows how to add the dependency for the Web framework:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>
The following example shows how to add the dependency for the WebFlux framework:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>
You can add the Actuator dependency as follows:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency>
![]() | Note |
---|---|
To run Spring Cloud Stream 2.0 apps in Cloud Foundry, you must add |
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:
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 |
---|---|
|
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>
.
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.
Whether the configuration inherits the environment of the application itself.
Default: true
.
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
.
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
.
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.
These properties are exposed via org.springframework.cloud.stream.config.BindingServiceProperties
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
.
0
to instanceCount - 1
.
Used for partitioning with RabbitMQ and with Kafka if autoRebalanceEnabled=false
.
Automatically set in Cloud Foundry to match the application’s instance index.A list of destinations that can be bound dynamically (for example, in a dynamic routing scenario). If set, only listed destinations can be bound.
Default: empty (letting any destination be bound).
The default binder to use, if multiple binders are configured. See Multiple Binders on the Classpath.
Default: empty.
This property is only applicable when the cloud
profile is active and Spring Cloud Connectors are provided with the application.
If the property is false
(the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors).
When set to true
, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.*
properties provided in the environment for the RabbitMQ binder).
The typical usage of this property is to be nested in a customized environment when connecting to multiple systems.
Default: false
.
The interval (in seconds) between retrying binding creation when, for example, the binder does not support late binding and the broker (for example, Apache Kafka) is down. Set it to zero to treat such conditions as fatal, preventing the application from starting.
Default: 30
Binding properties are supplied by using the format of spring.cloud.stream.bindings.<channelName>.<property>=<value>
.
The <channelName>
represents the name of the channel being configured (for example, output
for a Source
).
To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of spring.cloud.stream.default.<property>=<value>
.
When it comes to avoiding repetitions for extended binding properties, this format should be used - spring.cloud.stream.<binder-type>.default.<producer|consumer>.<property>=<value>
.
In what follows, we indicate where we have omitted the spring.cloud.stream.bindings.<channelName>.
prefix and focus just on the property name, with the understanding that the prefix ise included at runtime.
These properties are exposed via org.springframework.cloud.stream.config.BindingProperties
The following binding properties are available for both input and output bindings and must be prefixed with spring.cloud.stream.bindings.<channelName>.
(for example, spring.cloud.stream.bindings.input.destination=ticktock
).
Default values can be set by using the spring.cloud.stream.default
prefix (for example`spring.cloud.stream.default.contentType=application/json`).
String
values.
If not set, the channel name is used instead.
The default value of this property cannot be overridden.The consumer group of the channel. Applies only to inbound bindings. See Consumer Groups.
Default: null
(indicating an anonymous consumer).
The content type of the channel. See “Chapter 32, Content Type Negotiation”.
Default: application/json
.
The binder used by this binding. See “Section 30.4, “Multiple Binders on the Classpath”” for details.
Default: null
(the default binder is used, if it exists).
These properties are exposed via org.springframework.cloud.stream.binder.ConsumerProperties
The following binding properties are available for input bindings only and must be prefixed with spring.cloud.stream.bindings.<channelName>.consumer.
(for example, spring.cloud.stream.bindings.input.consumer.concurrency=3
).
Default values can be set by using the spring.cloud.stream.default.consumer
prefix (for example, spring.cloud.stream.default.consumer.headerMode=none
).
The concurrency of the inbound consumer.
Default: 1
.
Whether the consumer receives data from a partitioned producer.
Default: false
.
When set to none
, disables header parsing on input.
Effective only for messaging middleware that does not support message headers natively and requires header embedding.
This option is useful when consuming data from non-Spring Cloud Stream applications when native headers are not supported.
When set to headers
, it uses the middleware’s native header mechanism.
When set to embeddedHeaders
, it embeds headers into the message payload.
Default: depends on the binder implementation.
If processing fails, the number of attempts to process the message (including the first).
Set to 1
to disable retry.
Default: 3
.
The backoff initial interval on retry.
Default: 1000
.
The maximum backoff interval.
Default: 10000
.
The backoff multiplier.
Default: 2.0
.
Whether exceptions thrown by the listener that are not listed in the retryableExceptions
are retryable.
Default: true
.
When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from spring.cloud.stream.instanceIndex
).
When set to a negative value, it defaults to spring.cloud.stream.instanceIndex
.
See “Section 34.2, “Instance Index and Instance Count”” for more information.
Default: -1
.
When set to a value greater than equal to zero, it allows customizing the instance count of this consumer (if different from spring.cloud.stream.instanceCount
).
When set to a negative value, it defaults to spring.cloud.stream.instanceCount
.
See “Section 34.2, “Instance Index and Instance Count”” for more information.
Default: -1
.
A map of Throwable class names in the key and a boolean in the value.
Specify those exceptions (and subclasses) that will or won’t be retried.
Also see defaultRetriable
.
Example: spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false
.
Default: empty.
When set to true
, the inbound message is deserialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value deserializer).
When this configuration is being used, the inbound message unmarshalling is not based on the contentType
of the binding.
When native decoding is used, it is the responsibility of the producer to use an appropriate encoder (for example, the Kafka producer value serializer) to serialize the outbound message.
Also, when native encoding and decoding is used, the headerMode=embeddedHeaders
property is ignored and headers are not embedded in the message.
See the producer property useNativeEncoding
.
Default: false
.
These properties are exposed via org.springframework.cloud.stream.binder.ProducerProperties
The following binding properties are available for output bindings only and must be prefixed with spring.cloud.stream.bindings.<channelName>.producer.
(for example, spring.cloud.stream.bindings.input.producer.partitionKeyExpression=payload.id
).
Default values can be set by using the prefix spring.cloud.stream.default.producer
(for example, spring.cloud.stream.default.producer.partitionKeyExpression=payload.id
).
A SpEL expression that determines how to partition outbound data.
If set, or if partitionKeyExtractorClass
is set, outbound data on this channel is partitioned. partitionCount
must be set to a value greater than 1 to be effective.
Mutually exclusive with partitionKeyExtractorClass
.
See “Section 28.6, “Partitioning Support””.
Default: null.
A PartitionKeyExtractorStrategy
implementation.
If set, or if partitionKeyExpression
is set, outbound data on this channel is partitioned. partitionCount
must be set to a value greater than 1 to be effective.
Mutually exclusive with partitionKeyExpression
.
See “Section 28.6, “Partitioning Support””.
Default: null
.
A PartitionSelectorStrategy
implementation.
Mutually exclusive with partitionSelectorExpression
.
If neither is set, the partition is selected as the hashCode(key) % partitionCount
, where key
is computed through either partitionKeyExpression
or partitionKeyExtractorClass
.
Default: null
.
A SpEL expression for customizing partition selection.
Mutually exclusive with partitionSelectorClass
.
If neither is set, the partition is selected as the hashCode(key) % partitionCount
, where key
is computed through either partitionKeyExpression
or partitionKeyExtractorClass
.
Default: null
.
The number of target partitions for the data, if partitioning is enabled. Must be set to a value greater than 1 if the producer is partitioned. On Kafka, it is interpreted as a hint. The larger of this and the partition count of the target topic is used instead.
Default: 1
.
When set to none
, it disables header embedding on output.
It is effective only for messaging middleware that does not support message headers natively and requires header embedding.
This option is useful when producing data for non-Spring Cloud Stream applications when native headers are not supported.
When set to headers
, it uses the middleware’s native header mechanism.
When set to embeddedHeaders
, it embeds headers into the message payload.
Default: Depends on the binder implementation.
When set to true
, the outbound message is serialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value serializer).
When this configuration is being used, the outbound message marshalling is not based on the contentType
of the binding.
When native encoding is used, it is the responsibility of the consumer to use an appropriate decoder (for example, the Kafka consumer value de-serializer) to deserialize the inbound message.
Also, when native encoding and decoding is used, the headerMode=embeddedHeaders
property is ignored and headers are not embedded in the message.
See the consumer property useNativeDecoding
.
Default: false
.
When set to true
, if the binder supports asynchroous send results, send failures are sent to an error channel for the destination.
See “???” for more information.
Default: false
.
Besides the channels defined by using @EnableBinding
, Spring Cloud Stream lets applications send messages to dynamically bound destinations.
This is useful, for example, when the target destination needs to be determined at runtime.
Applications can do so by using the BinderAwareChannelResolver
bean, registered automatically by the @EnableBinding
annotation.
The 'spring.cloud.stream.dynamicDestinations' property can be used for restricting the dynamic destination names to a known set (whitelisting). If this property is not set, any destination can be bound dynamically.
The BinderAwareChannelResolver
can be used directly, as shown in the following example of a REST controller using a path variable to decide the target channel:
@EnableBinding @Controller public class SourceWithDynamicDestination { @Autowired private BinderAwareChannelResolver resolver; @RequestMapping(path = "/{target}", method = POST, consumes = "*/*") @ResponseStatus(HttpStatus.ACCEPTED) public void handleRequest(@RequestBody String body, @PathVariable("target") target, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) { sendMessage(body, target, contentType); } private void sendMessage(String body, String target, Object contentType) { resolver.resolveDestination(target).send(MessageBuilder.createMessage(body, new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType)))); } }
Now consider what happens when we start the application on the default port (8080) and make the following requests with CURL:
curl -H "Content-Type: application/json" -X POST -d "customer-1" http://localhost:8080/customers curl -H "Content-Type: application/json" -X POST -d "order-1" http://localhost:8080/orders
The destinations, 'customers' and 'orders', are created in the broker (in the exchange for Rabbit or in the topic for Kafka) with names of 'customers' and 'orders', and the data is published to the appropriate destinations.
The BinderAwareChannelResolver
is a general-purpose Spring Integration DestinationResolver
and can be injected in other components — for example, in a router using a SpEL expression based on the target
field of an incoming JSON message. The following example includes a router that reads SpEL expressions:
@EnableBinding @Controller public class SourceWithDynamicDestination { @Autowired private BinderAwareChannelResolver resolver; @RequestMapping(path = "/", method = POST, consumes = "application/json") @ResponseStatus(HttpStatus.ACCEPTED) public void handleRequest(@RequestBody String body, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) { sendMessage(body, contentType); } private void sendMessage(Object body, Object contentType) { routerChannel().send(MessageBuilder.createMessage(body, new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType)))); } @Bean(name = "routerChannel") public MessageChannel routerChannel() { return new DirectChannel(); } @Bean @ServiceActivator(inputChannel = "routerChannel") public ExpressionEvaluatingRouter router() { ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter(new SpelExpressionParser().parseExpression("payload.target")); router.setDefaultOutputChannelName("default-output"); router.setChannelResolver(resolver); return router; } }
The Router Sink Application uses this technique to create the destinations on-demand.
If the channel names are known in advance, you can configure the producer properties as with any other destination.
Alternatively, if you register a NewBindingCallback<>
bean, it is invoked just before the binding is created.
The callback takes the generic type of the extended producer properties used by the binder.
It has one method:
void configure(String channelName, MessageChannel channel, ProducerProperties producerProperties,
T extendedProducerProperties);
The following example shows how to use the RabbitMQ binder:
@Bean public NewBindingCallback<RabbitProducerProperties> dynamicConfigurer() { return (name, channel, props, extended) -> { props.setRequiredGroups("bindThisQueue"); extended.setQueueNameGroupOnly(true); extended.setAutoBindDlq(true); extended.setDeadLetterQueueName("myDLQ"); }; }
![]() | Note |
---|---|
If you need to support dynamic destinations with multiple binder types, use |
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:
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 |
---|---|
As a supplement to the details to follow, you may also want to read the following blog post. |
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 |
---|---|
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.
In order for the framework to succeed in passing the incoming Message
as an argument to this handler, it has to somehow transform the payload of the Message
type from the wire format to a Person
type.
In other words, the framework must locate and apply the appropriate MessageConverter
.
To accomplish that, the framework needs some instructions from the user.
One of these instructions is already provided by the signature of the handler method itself (Person
type).
Consequently, in theory, that should be (and, in some cases, is) enough.
However, for the majority of use cases, in order to select the appropriate MessageConverter
, the framework needs an additional piece of information.
That missing piece is contentType
.
Spring Cloud Stream provides three mechanisms to define contentType
(in order of precedence):
contentType
can be communicated through the Message itself. By providing a contentType
header, you declare the content type to use to locate and apply the appropriate MessageConverter
.BINDING: The contentType
can be set per destination binding by setting the spring.cloud.stream.bindings.input.content-type
property.
![]() | Note |
---|---|
The |
contentType
is not present in the Message
header or the binding, the default application/json
content type is used to
locate and apply the appropriate MessageConverter
.As mentioned earlier, the preceding list also demonstrates the order of precedence in case of a tie. For example, a header-provided content type takes precedence over any other content type. The same applies for a content type set on a per-binding basis, which essentially lets you override the default content type. However, it also provides a sensible default (which was determined from community feedback).
Another reason for making application/json
the default stems from the interoperability requirements driven by distributed microservices architectures, where producer and consumer not only run in different JVMs but can also run on different non-JVM platforms.
When the non-void handler method returns, if the the return value is already a Message
, that Message
becomes the payload. However, when the return value is not a Message
, the new Message
is constructed with the return value as the payload while inheriting
headers from the input Message
minus the headers defined or filtered by SpringIntegrationProperties.messageHandlerNotPropagatedHeaders
.
By default, there is only one header set there: contentType
. This means that the new Message
does not have contentType
header set, thus ensuring that the contentType
can evolve.
You can always opt out of returning a Message
from the handler method where you can inject any header you wish.
If there is an internal pipeline, the Message
is sent to the next handler by going through the same process of conversion. However, if there is no internal pipeline or you have reached the end of it, the Message
is sent back to the output destination.
As mentioned earlier, for the framework to select the appropriate MessageConverter
, it requires argument type and, optionally, content type information.
The logic for selecting the appropriate MessageConverter
resides with the argument resolvers (HandlerMethodArgumentResolvers
), which trigger right before the invocation of the user-defined handler method (which is when the actual argument type is known to the framework).
If the argument type does not match the type of the current payload, the framework delegates to the stack of the
pre-configured MessageConverters
to see if any one of them can convert the payload.
As you can see, the Object fromMessage(Message<?> message, Class<?> targetClass);
operation of the MessageConverter takes targetClass
as one of its arguments.
The framework also ensures that the provided Message
always contains a contentType
header.
When no contentType header was already present, it injects either the per-binding contentType
header or the default contentType
header.
The combination of contentType
argument type is the mechanism by which framework determines if message can be converted to a target type.
If no appropriate MessageConverter
is found, an exception is thrown, which you can handle by adding a custom MessageConverter
(see “Section 32.3, “User-defined Message Converters””).
But what if the payload type matches the target type declared by the handler method? In this case, there is nothing to convert, and the
payload is passed unmodified. While this sounds pretty straightforward and logical, keep in mind handler methods that take a Message<?>
or Object
as an argument.
By declaring the target type to be Object
(which is an instanceof
everything in Java), you essentially forfeit the conversion process.
![]() | Note |
---|---|
Do not expect |
MessageConverters
define two methods:
Object fromMessage(Message<?> message, Class<?> targetClass);
Message<?> toMessage(Object payload, @Nullable MessageHeaders headers);
It is important to understand the contract of these methods and their usage, specifically in the context of Spring Cloud Stream.
The fromMessage
method converts an incoming Message
to an argument type.
The payload of the Message
could be any type, and it is
up to the actual implementation of the MessageConverter
to support multiple types.
For example, some JSON converter may support the payload type as byte[]
, String
, and others.
This is important when the application contains an internal pipeline (that is, input → handler1 → handler2 →. . . → output) and the output of the upstream handler results in a Message
which may not be in the initial wire format.
However, the toMessage
method has a more strict contract and must always convert Message
to the wire format: byte[]
.
So, for all intents and purposes (and especially when implementing your own converter) you regard the two methods as having the following signatures:
Object fromMessage(Message<?> message, Class<?> targetClass); Message<byte[]> toMessage(Object payload, @Nullable MessageHeaders headers);
As mentioned earlier, the framework already provides a stack of MessageConverters
to handle most common use cases.
The following list describes the provided MessageConverters
, in order of precedence (the first MessageConverter
that works is used):
ApplicationJsonMessageMarshallingConverter
: Variation of the org.springframework.messaging.converter.MappingJackson2MessageConverter
. Supports conversion of the payload of the Message
to/from POJO for cases when contentType
is application/json
(DEFAULT).TupleJsonMessageConverter
: DEPRECATED Supports conversion of the payload of the Message
to/from org.springframework.tuple.Tuple
.ByteArrayMessageConverter
: Supports conversion of the payload of the Message
from byte[]
to byte[]
for cases when contentType
is application/octet-stream
. It is essentially a pass through and exists primarily for backward compatibility.ObjectStringMessageConverter
: Supports conversion of any type to a String
when contentType
is text/plain
.
It invokes Object’s toString()
method or, if the payload is byte[]
, a new String(byte[])
.JavaSerializationMessageConverter
: DEPRECATED Supports conversion based on java serialization when contentType
is application/x-java-serialized-object
.KryoMessageConverter
: DEPRECATED Supports conversion based on Kryo serialization when contentType
is application/x-java-object
.JsonUnmarshallingConverter
: Similar to the ApplicationJsonMessageMarshallingConverter
. It supports conversion of any type when contentType
is application/x-java-object
.
It expects the actual type information to be embedded in the contentType
as an attribute (for example, application/x-java-object;type=foo.bar.Cat
).When no appropriate converter is found, the framework throws an exception. When that happens, you should check your code and configuration and ensure you did not miss anything (that is, ensure that you provided a contentType
by using a binding or a header).
However, most likely, you found some uncommon case (such as a custom contentType
perhaps) and the current stack of provided MessageConverters
does not know how to convert. If that is the case, you can add custom MessageConverter
. See Section 32.3, “User-defined Message Converters”.
Spring Cloud Stream exposes a mechanism to define and register additional MessageConverters
.
To use it, implement org.springframework.messaging.converter.MessageConverter
, configure it as a @Bean
, and annotate it with @StreamMessageConverter
.
It is then apended to the existing stack of `MessageConverter`s.
![]() | Note |
---|---|
It is important to understand that custom |
The following example shows how to create a message converter bean to support a new content type called application/bar
:
@EnableBinding(Sink.class) @SpringBootApplication public static class SinkApplication { ... @Bean @StreamMessageConverter public MessageConverter customMessageConverter() { return new MyCustomMessageConverter(); } } public class MyCustomMessageConverter extends AbstractMessageConverter { public MyCustomMessageConverter() { super(new MimeType("application", "bar")); } @Override protected boolean supports(Class<?> clazz) { return (Bar.class.equals(clazz)); } @Override protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) { Object payload = message.getPayload(); return (payload instanceof Bar ? payload : new Bar((byte[]) payload)); } }
Spring Cloud Stream also provides support for Avro-based converters and schema evolution. See “Chapter 33, Schema Evolution Support” for details.
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. However, many applications benefit from having access to an explicit schema that describes the binary data format. A schema registry lets you store schema information in a textual format (typically JSON) and makes that information accessible to various applications that need it to receive and send data in binary format. A schema is referenceable as a tuple consisting of:
This following sections goes through the details of various components involved in schema evolution process.
The client-side abstraction for interacting with schema registry servers is the SchemaRegistryClient
interface, which has the following structure:
public interface SchemaRegistryClient { SchemaRegistrationResponse register(String subject, String format, String schema); String fetch(SchemaReference schemaReference); String fetch(Integer id); }
Spring Cloud Stream provides out-of-the-box implementations for interacting with its own schema server and for interacting with the Confluent Schema Registry.
A client for the Spring Cloud Stream schema registry can be configured by using the @EnableSchemaRegistryClient
, as follows:
@EnableBinding(Sink.class) @SpringBootApplication @EnableSchemaRegistryClient public static class AvroSinkApplication { ... }
![]() | Note |
---|---|
The default converter is optimized to cache not only the schemas from the remote server but also the |
The Schema Registry Client supports the following properties:
spring.cloud.stream.schemaRegistryClient.endpoint
http
or https
) , port, and context path.http://localhost:8990/
spring.cloud.stream.schemaRegistryClient.cached
false
, as the caching happens in the message converter.
Clients using the schema registry client should set this to true
.true
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
.
If an identical schema is already found, then a reference to it is retrieved.
If not, the schema is registered, and a new version number is provided.
The message is sent with a contentType
header by using the following scheme: application/[prefix].[subject].v[version]+avro
, where prefix
is configurable and subject
is deduced from the payload type.
For example, a message of the type User
might be sent as a binary payload with a content type of application/vnd.user.v2+avro
, where user
is the subject and 2
is the version number.
When receiving messages, the converter infers the schema reference from the header of the incoming message and tries to retrieve it. The schema is used as the writer schema in the deserialization process.
If you have enabled Avro based schema registry client by setting spring.cloud.stream.bindings.output.contentType=application/*+avro
, you can customize the behavior of the registration by setting the following properties.
Enable if you want the converter to use reflection to infer a Schema from a POJO.
Default: false
null
Registers any .avsc
files listed in this property with the Schema Server.
Default: empty
The prefix to be used on the Content-Type header.
Default: vnd
Spring Cloud Stream provides support for schema-based message converters through its spring-cloud-stream-schema
module.
Currently, the only serialization format supported out of the box for schema-based message converters is Apache Avro, with more formats to be added in future versions.
The spring-cloud-stream-schema
module contains two types of message converters that can be used for Apache Avro serialization:
The AvroSchemaMessageConverter
supports serializing and deserializing messages either by using a predefined schema or by using the schema information available in the class (either reflectively or contained in the SpecificRecord
).
If you provide a custom converter, then the default AvroSchemaMessageConverter bean is not created. The following example shows a custom converter:
To use custom converters, you can simply add it to the application context, optionally specifying one or more MimeTypes
with which to associate it.
The default MimeType
is application/avro
.
If the target type of the conversion is a GenericRecord
, a schema must be set.
The following example shows how to configure a converter in a sink application by registering the Apache Avro MessageConverter
without a predefined schema.
In this example, note that the mime type value is avro/bytes
, not the default application/avro
.
@EnableBinding(Sink.class) @SpringBootApplication public static class SinkApplication { ... @Bean public MessageConverter userMessageConverter() { return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")); } }
Conversely, the following application registers a converter with a predefined schema (found on the classpath):
@EnableBinding(Sink.class) @SpringBootApplication public static class SinkApplication { ... @Bean public MessageConverter userMessageConverter() { AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")); converter.setSchemaLocation(new ClassPathResource("schemas/User.avro")); return converter; } }
Spring Cloud Stream provides a schema registry server implementation.
To use it, you can add the spring-cloud-stream-schema-server
artifact to your project and use the @EnableSchemaRegistryServer
annotation, which adds the schema registry server REST controller to your application.
This annotation is intended to be used with Spring Boot web applications, and the listening port of the server is controlled by the server.port
property.
The spring.cloud.stream.schema.server.path
property can be used to control the root path of the schema server (especially when it is embedded in other applications).
The spring.cloud.stream.schema.server.allowSchemaDeletion
boolean property enables the deletion of a schema. By default, this is disabled.
The schema registry server uses a relational database to store the schemas. By default, it uses an embedded database. You can customize the schema storage by using the Spring Boot SQL database and JDBC configuration options.
The following example shows a Spring Boot application that enables the schema registry:
@SpringBootApplication @EnableSchemaRegistryServer public class SchemaRegistryServerApplication { public static void main(String[] args) { SpringApplication.run(SchemaRegistryServerApplication.class, args); } }
The Schema Registry Server API consists of the following operations:
POST /
— see “the section called “Registering a New Schema””GET /{subject}/{format}
— see “the section called “Retrieving an Existing Schema by Subject and Format””GET /schemas/{id}
— see “the section called “Retrieving an Existing Schema by ID””DELETE /{subject}/{format}/{version}
— see “the section called “Deleting a Schema by Subject, Format, and Version””DELETE /schemas/{id}
— see “the section called “Deleting a Schema by ID””DELETE /{subject}
— see “the section called “Deleting a Schema by Subject””To register a new schema, send a POST
request to the /
endpoint.
The /
accepts a JSON payload with the following fields:
subject
: The schema subjectformat
: The schema formatdefinition
: The schema definitionIts response is a schema object in JSON, with the following fields:
id
: The schema IDsubject
: The schema subjectformat
: The schema formatversion
: The schema versiondefinition
: The schema definitionTo retrieve an existing schema by subject, format, and version, send GET
request to the /{subject}/{format}/{version}
endpoint.
Its response is a schema object in JSON, with the following fields:
id
: The schema IDsubject
: The schema subjectformat
: The schema formatversion
: The schema versiondefinition
: The schema definitionTo retrieve an existing schema by subject and format, send a GET
request to the /subject/format
endpoint.
Its response is a list of schemas with each schema object in JSON, with the following fields:
id
: The schema IDsubject
: The schema subjectformat
: The schema formatversion
: The schema versiondefinition
: The schema definitionTo retrieve a schema by its ID, send a GET
request to the /schemas/{id}
endpoint.
Its response is a schema object in JSON, with the following fields:
id
: The schema IDsubject
: The schema subjectformat
: The schema formatversion
: The schema versiondefinition
: The schema definitionTo delete a schema identified by its subject, format, and version, send a DELETE
request to the /{subject}/{format}/{version}
endpoint.
To delete a schema by its ID, send a DELETE
request to the /schemas/{id}
endpoint.
DELETE /{subject}
Delete existing schemas by their subject.
![]() | Note |
---|---|
This note applies to users of Spring Cloud Stream 1.1.0.RELEASE only.
Spring Cloud Stream 1.1.0.RELEASE used the table name, |
The default configuration creates a DefaultSchemaRegistryClient
bean.
If you want to use the Confluent schema registry, you need to create a bean of type ConfluentSchemaRegistryClient
, which supersedes the one configured by default by the framework. The following example shows how to create such a bean:
@Bean public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){ ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(); client.setEndpoint(endpoint); return client; }
![]() | Note |
---|---|
The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0. |
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:
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).
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.
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
.
When reading messages that contain version information (that is, a contentType
header with a scheme like the one described under “Section 33.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).
![]() | 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 |
Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:
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
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.
By default, spring.cloud.stream.instanceCount
is 1
, and spring.cloud.stream.instanceIndex
is 0
.
In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (for example, the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances.
Partitioning in Spring Cloud Stream consists of two tasks:
You can configure an output binding to send partitioned data by setting one and only one of its partitionKeyExpression
or partitionKeyExtractorName
properties, as well as its partitionCount
property.
For example, the following is a valid and typical configuration:
spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload.id spring.cloud.stream.bindings.output.producer.partitionCount=5
Based on that example configuration, data is sent to the target partition by using the following logic.
A partition key’s value is calculated for each message sent to a partitioned output channel based on the partitionKeyExpression
.
The partitionKeyExpression
is a SpEL expression that is evaluated against the outbound message for extracting the partitioning key.
If a SpEL expression is not sufficient for your needs, you can instead calculate the partition key value by providing an implementation of org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy
and configuring it as a bean (by using the @Bean
annotation).
If you have more then one bean of type org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy
available in the Application Context, you can further filter it by specifying its name with the partitionKeyExtractorName
property, as shown in the following example:
--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor --spring.cloud.stream.bindings.output.producer.partitionCount=5 . . . @Bean public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() { return new CustomPartitionKeyExtractorClass(); }
![]() | Note |
---|---|
In previous versions of Spring Cloud Stream, you could specify the implementation of |
Once the message key is calculated, the partition selection process determines the target partition as a value between 0
and partitionCount - 1
.
The default calculation, applicable in most scenarios, is based on the following formula: key.hashCode() % partitionCount
.
This can be customized on the binding, either by setting a SpEL expression to be evaluated against the 'key' (through the partitionSelectorExpression
property) or by configuring an implementation of org.springframework.cloud.stream.binder.PartitionSelectorStrategy
as a bean (by using the @Bean annotation).
Similar to the PartitionKeyExtractorStrategy
, you can further filter it by using the spring.cloud.stream.bindings.output.producer.partitionSelectorName
property when more than one bean of this type is available in the Application Context, as shown in the following example:
--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector . . . @Bean public CustomPartitionSelectorClass customPartitionSelector() { return new CustomPartitionSelectorClass(); }
![]() | Note |
---|---|
In previous versions of Spring Cloud Stream you could specify the implementation of |
An input binding (with the channel name input
) is configured to receive partitioned data by setting its partitioned
property, as well as the instanceIndex
and instanceCount
properties on the application itself, as shown in the following example:
spring.cloud.stream.bindings.input.consumer.partitioned=true spring.cloud.stream.instanceIndex=3 spring.cloud.stream.instanceCount=5
The instanceCount
value represents the total number of application instances between which the data should be partitioned.
The instanceIndex
must be a unique value across the multiple instances, with a value between 0
and instanceCount - 1
.
The instance index helps each application instance to identify the unique partition(s) from which it receives data.
It is required by binders using technology that does not support partitioning natively.
For example, with RabbitMQ, there is a queue for each partition, with the queue name containing the instance index.
With Kafka, if autoRebalanceEnabled
is true
(default), Kafka takes care of distributing partitions across instances, and these properties are not required.
If 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.
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> <scope>test</scope> </dependency>
![]() | Note |
---|---|
The |
The TestSupportBinder
lets you interact with the bound channels and inspect any messages sent and received by the application.
For outbound message channels, the TestSupportBinder
registers a single subscriber and retains the messages emitted by the application in a MessageCollector
.
They can be retrieved during tests and have assertions made against them.
You can also send messages to inbound message channels so that the consumer application can consume the messages. The following example shows how to test both input and output channels on a processor:
@RunWith(SpringRunner.class) @SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.RANDOM_PORT) public class ExampleTest { @Autowired private Processor processor; @Autowired private MessageCollector messageCollector; @Test @SuppressWarnings("unchecked") public void testWiring() { Message<String> message = new GenericMessage<>("hello"); processor.input().send(message); Message<String> received = (Message<String>) messageCollector.forChannel(processor.output()).poll(); assertThat(received.getPayload(), equalTo("hello world")); } @SpringBootApplication @EnableBinding(Processor.class) public static class MyProcessor { @Autowired private Processor channels; @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public String transform(String in) { return in + " world"; } } }
In the preceding example, we create an application that has an input channel and an output channel, both bound through the Processor
interface.
The bound interface is injected into the test so that we can have access to both channels.
We send a message on the input channel, and we use the MessageCollector
provided by Spring Cloud Stream’s test support to capture that the message has been sent to the output channel as a result.
Once we have received the message, we can validate that the component functions correctly.
The intent behind the test binder superseding all the other binders on the classpath is to make it easy to test your applications without making changes to your production dependencies.
In some cases (for example, integration tests) it is useful to use the actual production binders instead, and that requires disabling the test binder autoconfiguration.
To do so, you can exclude the org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration
class by using one of the Spring Boot autoconfiguration exclusion mechanisms, as shown in the following example:
@SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class) @EnableBinding(Processor.class) public static class MyProcessor { @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) 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
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.
By default management.health.binders.enabled
is set to false
.
Setting management.health.binders.enabled
to true
enables the health indicator, allowing you to access the /health
endpoint to retrieve the binder health indicators.
Health indicators are binder-specific and certain binder implementations may not necessarily provide a health indicator.
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:
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}}}
Allows white listing application properties that are added to the metrics payload
Default: null.
Pattern to control the 'meters' one wants to capture.
For example, specifying spring.integration.*
captures metric information for meters whose name starts with spring.integration.
Default: all 'meters' are captured.
Interval to control the rate of publishing metric data.
Default: 1 min
Consider the following:
java -jar time-source.jar \ --spring.cloud.stream.bindings.applicationMetrics.destination=someMetrics \ --spring.cloud.stream.metrics.properties=spring.application** \ --spring.cloud.stream.metrics.meter-filter=spring.integration.*
The following example shows the payload of the data published to the binding destination as a result of the preceding command:
{ "name": "application", "createdTime": "2018-03-23T14:48:12.700Z", "properties": { }, "metrics": [ { "id": { "name": "spring.integration.send", "tags": [ { "key": "exception", "value": "none" }, { "key": "name", "value": "input" }, { "key": "result", "value": "success" }, { "key": "type", "value": "channel" } ], "type": "TIMER", "description": "Send processing time", "baseUnit": "milliseconds" }, "timestamp": "2018-03-23T14:48:12.697Z", "sum": 130.340546, "count": 6, "mean": 21.72342433333333, "upper": 116.176299, "total": 130.340546 } ] }
![]() | Note |
---|---|
Given that the format of the Metric message has slightly changed after migrating to Micrometer, the published message will also have
a |
For Spring Cloud Stream samples, see the spring-cloud-stream-samples repository on GitHub.
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.
To use Apache Kafka binder, you need to add spring-cloud-stream-binder-kafka
as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-kafka</artifactId> </dependency>
Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown inn the following example for Maven:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-kafka</artifactId> </dependency>
The following image shows a simplified diagram of how the Apache Kafka binder operates:
The Apache Kafka Binder implementation maps each destination to an Apache Kafka topic. The consumer group maps directly to the same Apache Kafka concept. Partitioning also maps directly to Apache Kafka partitions as well.
The binder currently uses the Apache Kafka kafka-clients
1.0.0 jar and is designed to be used with a broker of at least that version.
This client can communicate with older brokers (see the Kafka documentation), but certain features may not be available.
For example, with versions earlier than 0.11.x.x, native headers are not supported.
Also, 0.11.x.x does not support the autoAddPartitions
property.
This section contains the configuration options used by the Apache Kafka binder.
For common configuration options and properties pertaining to binder, see the core documentation.
A list of brokers to which the Kafka binder connects.
Default: localhost
.
brokers
allows hosts specified with or without port information (for example, host1,host2:port2
).
This sets the default port when no port is configured in the broker list.
Default: 9092
.
Key/Value map of client properties (both producers and consumer) passed to all clients created by the binder. Due to the fact that these properties are used by both producers and consumers, usage should be restricted to common properties — for example, security settings. Properties here supersede any properties set in boot.
Default: Empty map.
Key/Value map of arbitrary Kafka client consumer properties.
Properties here supersede any properties set in boot and in the configuration
property above.
Default: Empty map.
The list of custom headers that are transported by the binder.
Only required when communicating with older applications (⇐ 1.3.x) with a kafka-clients
version < 0.11.0.0. Newer versions support headers natively.
Default: empty.
The time to wait to get partition information, in seconds. Health reports as down if this timer expires.
Default: 10.
The number of required acks on the broker.
See the Kafka documentation for the producer acks
property.
Default: 1
.
Effective only if autoCreateTopics
or autoAddPartitions
is set.
The global minimum number of partitions that the binder configures on topics on which it produces or consumes data.
It can be superseded by the partitionCount
setting of the producer or by the value of instanceCount * concurrency
settings of the producer (if either is larger).
Default: 1
.
Key/Value map of arbitrary Kafka client producer properties.
Properties here supersede any properties set in boot and in the configuration
property above.
Default: Empty map.
The replication factor of auto-created topics if autoCreateTopics
is active.
Can be overridden on each binding.
Default: 1
.
If set to true
, the binder creates new topics automatically.
If set to false
, the binder relies on the topics being already configured.
In the latter case, if the topics do not exist, the binder fails to start.
![]() | Note |
---|---|
This setting is independent of the |
Default: true
.
If set to true
, the binder creates new partitions if required.
If set to false
, the binder relies on the partition size of the topic being already configured.
If the partition count of the target topic is smaller than the expected value, the binder fails to start.
Default: false
.
Enables transactions in the binder. See transaction.id
in the Kafka documentation and Transactions in the spring-kafka
documentation.
When transactions are enabled, individual producer
properties are ignored and all producers use the spring.cloud.stream.kafka.binder.transaction.producer.*
properties.
Default null
(no transactions)
Global producer properties for producers in a transactional binder.
See spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix
and Section 39.3.3, “Kafka Producer Properties” and the general producer properties supported by all binders.
Default: See individual producer properties.
The bean name of a KafkaHeaderMapper
used for mapping spring-messaging
headers to and from Kafka headers.
Use this, for example, if you wish to customize the trusted packages in a DefaultKafkaHeaderMapper
that uses JSON deserialization for the headers.
Default: none.
The following properties are available for Kafka consumers only and
must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.consumer.
.
A Map
of Kafka topic properties used when provisioning topics — for example, spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0
Default: none.
A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
Used when provisioning new topics.
See the NewTopic
Javadocs in the kafka-clients
jar.
Default: none.
The replication factor to use when provisioning topics. Overrides the binder-wide setting.
Ignored if replicas-assignments
is present.
Default: none (the binder-wide default of 1 is used).
When true
, topic partitions is automatically rebalanced between the members of a consumer group.
When false
, each consumer is assigned a fixed set of partitions based on spring.cloud.stream.instanceCount
and spring.cloud.stream.instanceIndex
.
This requires both the spring.cloud.stream.instanceCount
and spring.cloud.stream.instanceIndex
properties to be set appropriately on each launched instance.
The value of the spring.cloud.stream.instanceCount
property must typically be greater than 1 in this case.
Default: true
.
When autoCommitOffset
is true
, this setting dictates whether to commit the offset after each record is processed.
By default, offsets are committed after all records in the batch of records returned by consumer.poll()
have been processed.
The number of records returned by a poll can be controlled with the max.poll.records
Kafka property, which is set through the consumer configuration
property.
Setting this to true
may cause a degradation in performance, but doing so reduces the likelihood of redelivered records when a failure occurs.
Also, see the binder requiredAcks
property, which also affects the performance of committing offsets.
Default: false
.
Whether to autocommit offsets when a message has been processed.
If set to false
, a header with the key kafka_acknowledgment
of the type org.springframework.kafka.support.Acknowledgment
header is present in the inbound message.
Applications may use this header for acknowledging messages.
See the examples section for details.
When this property is set to false
, Kafka binder sets the ack mode to org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL
and the application is responsible for acknowledging records.
Also see ackEachRecord
.
Default: true
.
Effective only if autoCommitOffset
is set to true
.
If set to false
, it suppresses auto-commits for messages that result in errors and commits only for successful messages. It allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.
If set to true
, it always auto-commits (if auto-commit is enabled).
If not set (the default), it effectively has the same value as enableDlq
, auto-committing erroneous messages if they are sent to a DLQ and not committing them otherwise.
Default: not set.
Whether to reset offsets on the consumer to the value provided by startOffset.
Default: false
.
The starting offset for new groups.
Allowed values: earliest
and latest
.
If the consumer group is set explicitly for the consumer 'binding' (through spring.cloud.stream.bindings.<channelName>.group
), 'startOffset' is set to earliest
. Otherwise, it is set to latest
for the anonymous
consumer group.
Also see resetOffsets
(earlier in this list).
Default: null (equivalent to earliest
).
When set to true, it enables DLQ behavior for the consumer.
By default, messages that result in errors are forwarded to a topic named error.<destination>.<group>
.
The DLQ topic name can be configurable by setting the dlqName
property.
This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
See Section 39.6, “Dead-Letter Topic Processing” processing for more information.
Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: x-original-topic
, x-exception-message
, and x-exception-stacktrace
as byte[]
.
Not allowed when destinationIsPattern
is true
.
Default: false
.
Map with a key/value pair containing generic Kafka consumer properties.
Default: Empty map.
The name of the DLQ topic to receive the error messages.
Default: null (If not specified, messages that result in errors are forwarded to a topic named error.<destination>.<group>
).
Using this, DLQ-specific producer properties can be set. All the properties available through kafka producer properties can be set through this property.
Default: Default Kafka producer properties.
Indicates which standard headers are populated by the inbound channel adapter.
Allowed values: none
, id
, timestamp
, or both
.
Useful if using native deserialization and the first component to receive a message needs an id
(such as an aggregator that is configured to use a JDBC message store).
Default: none
The name of a bean that implements RecordMessageConverter
. Used in the inbound channel adapter to replace the default MessagingMessageConverter
.
Default: null
The interval, in milliseconds, between events indicating that no messages have recently been received.
Use an ApplicationListener<ListenerContainerIdleEvent>
to receive these events.
See the section called “Example: Pausing and Resuming the Consumer” for a usage example.
Default: 30000
When true, the destination is treated as a regular expression Pattern
used to match topic names by the broker.
When true, topics are not provisioned, and enableDlq
is not allowed, because the binder does not know the topic names during the provisioning phase.
Note, the time taken to detect new topics that match the pattern is controlled by the consumer property metadata.max.age.ms
, which (at the time of writing) defaults to 300,000ms (5 minutes).
This can be configured using the configuration
property above.
Default: false
The following properties are available for Kafka producers only and
must be prefixed with spring.cloud.stream.kafka.bindings.<channelName>.producer.
.
A Map
of Kafka topic properties used when provisioning new topics — for example, spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0
Default: none.
A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
Used when provisioning new topics.
See NewTopic
javadocs in the kafka-clients
jar.
Default: none.
The replication factor to use when provisioning new topics. Overrides the binder-wide setting.
Ignored if replicas-assignments
is present.
Default: none (the binder-wide default of 1 is used).
Upper limit, in bytes, of how much data the Kafka producer attempts to batch before sending.
Default: 16384
.
Whether the producer is synchronous.
Default: false
.
How long the producer waits to allow more messages to accumulate in the same batch before sending the messages. (Normally, the producer does not wait at all and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency.
Default: 0
.
A SpEL expression evaluated against the outgoing message used to populate the key of the produced Kafka message — for example, headers['myKey']
.
The payload cannot be used because, by the time this expression is evaluated, the payload is already in the form of a byte[]
.
Default: none
.
A comma-delimited list of simple patterns to match Spring messaging headers to be mapped to the Kafka Headers
in the ProducerRecord
.
Patterns can begin or end with the wildcard character (asterisk).
Patterns can be negated by prefixing with !
.
Matching stops after the first match (positive or negative).
For example !ask,as*
will pass ash
but not ask
.
id
and timestamp
are never mapped.
Default: *
(all headers - except the id
and timestamp
)
Map with a key/value pair containing generic Kafka producer properties.
Default: Empty map.
![]() | Note |
---|---|
The Kafka binder uses the |
In this section, we show the use of the preceding properties for specific scenarios.
This example illustrates how one may manually acknowledge offsets in a consumer application.
This example requires that spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset
be set to false
.
Use the corresponding input channel name for your example.
@SpringBootApplication @EnableBinding(Sink.class) public class ManuallyAcknowdledgingConsumer { public static void main(String[] args) { SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); } @StreamListener(Sink.INPUT) public void process(Message<?> message) { Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); if (acknowledgment != null) { System.out.println("Acknowledgment provided"); acknowledgment.acknowledge(); } } }
Apache Kafka 0.9 supports secure connections between client and brokers.
To take advantage of this feature, follow the guidelines in the Apache Kafka Documentation as well as the Kafka 0.9 security guidelines from the Confluent documentation.
Use the spring.cloud.stream.kafka.binder.configuration
option to set security properties for all clients created by the binder.
For example, to set security.protocol
to SASL_SSL
, set the following property:
spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL
All the other security properties can be set in a similar manner.
When using Kerberos, follow the instructions in the reference documentation for creating and referencing the JAAS configuration.
Spring Cloud Stream supports passing JAAS configuration information to the application by using a JAAS configuration file and using Spring Boot properties.
The JAAS and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties. The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using a JAAS configuration file:
java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \
--spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
--spring.cloud.stream.bindings.input.destination=stream.ticktock \
--spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT
As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications by using Spring Boot properties.
The following properties can be used to configure the login context of the Kafka client:
The login module name. Not necessary to be set in normal cases.
Default: com.sun.security.auth.module.Krb5LoginModule
.
The control flag of the login module.
Default: required
.
Map with a key/value pair containing the login module options.
Default: Empty map.
The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using Spring Boot configuration properties:
java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ --spring.cloud.stream.bindings.input.destination=stream.ticktock \ --spring.cloud.stream.kafka.binder.autoCreateTopics=false \ --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \ --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \ --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \ --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \ --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM
The preceding example represents the equivalent of the following JAAS file:
KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/etc/security/keytabs/kafka_client.keytab" principal="[email protected]"; };
If the topics required already exist on the broker or will be created by an administrator, autocreation can be turned off and only client JAAS properties need to be sent.
![]() | Note |
---|---|
Do not mix JAAS configuration files and Spring Boot properties in the same application.
If the |
![]() | Note |
---|---|
Be careful when using the |
If you wish to suspend consumption but not cause a partition rebalance, you can pause and resume the consumer.
This is facilitated by adding the Consumer
as a parameter to your @StreamListener
.
To resume, you need an ApplicationListener
for ListenerContainerIdleEvent
instances.
The frequency at which events are published is controlled by the idleEventInterval
property.
Since the consumer is not thread-safe, you must call these methods on the calling thread.
The following simple application shows how to pause and resume:
@SpringBootApplication @EnableBinding(Sink.class) public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } @StreamListener(Sink.INPUT) public void in(String in, @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) { System.out.println(in); consumer.pause(Collections.singleton(new TopicPartition("myTopic", 0))); } @Bean public ApplicationListener<ListenerContainerIdleEvent> idleListener() { return event -> { System.out.println(event); if (event.getConsumer().paused().size() > 0) { event.getConsumer().resume(event.getConsumer().paused()); } }; } }
Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel. See Section 29.4, “Error Handling” for more information.
The payload of the ErrorMessage
for a send failure is a KafkaSendFailureException
with properties:
failedMessage
: The Spring Messaging Message<?>
that failed to be sent.record
: The raw ProducerRecord
that was created from the failedMessage
There is no automatic handling of producer exceptions (such as sending to a Dead-Letter queue). You can consume these exceptions with your own Spring Integration flow.
Kafka binder module exposes the following metrics:
spring.cloud.stream.binder.kafka.offset
: This metric indicates how many messages have not been yet consumed from a given binder’s topic by a given consumer group.
The metrics provided are based on the Mircometer metrics library. The metric contains the consumer group information, topic and the actual lag in committed offset from the latest offset on the topic.
This metric is particularly useful for providing auto-scaling feedback to a PaaS platform.
Because you cannot anticipate how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. If the reason for the dead-lettering is transient, you may wish to route the messages back to the original topic. However, if the problem is a permanent issue, that could cause an infinite loop. The sample Spring Boot application within this topic is an example of how to route those messages back to the original topic, but it moves them to a “parking lot” topic after three attempts. The application is another spring-cloud-stream application that reads from the dead-letter topic. It terminates when no messages are received for 5 seconds.
The examples assume the original destination is so8400out
and the consumer group is so8400
.
There are a couple of strategies to consider:
The following code listings show the sample application:
application.properties.
spring.cloud.stream.bindings.input.group=so8400replay spring.cloud.stream.bindings.input.destination=error.so8400out.so8400 spring.cloud.stream.bindings.output.destination=so8400out spring.cloud.stream.bindings.output.producer.partitioned=true spring.cloud.stream.bindings.parkingLot.destination=so8400in.parkingLot spring.cloud.stream.bindings.parkingLot.producer.partitioned=true spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=earliest spring.cloud.stream.kafka.binder.headers=x-retries
Application.
@SpringBootApplication @EnableBinding(TwoOutputProcessor.class) public class ReRouteDlqKApplication implements CommandLineRunner { private static final String X_RETRIES_HEADER = "x-retries"; public static void main(String[] args) { SpringApplication.run(ReRouteDlqKApplication.class, args).close(); } private final AtomicInteger processed = new AtomicInteger(); @Autowired private MessageChannel parkingLot; @StreamListener(Processor.INPUT) @SendTo(Processor.OUTPUT) public Message<?> reRoute(Message<?> failed) { processed.incrementAndGet(); Integer retries = failed.getHeaders().get(X_RETRIES_HEADER, Integer.class); if (retries == null) { System.out.println("First retry for " + failed); return MessageBuilder.fromMessage(failed) .setHeader(X_RETRIES_HEADER, new Integer(1)) .setHeader(BinderHeaders.PARTITION_OVERRIDE, failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)) .build(); } else if (retries.intValue() < 3) { System.out.println("Another retry for " + failed); return MessageBuilder.fromMessage(failed) .setHeader(X_RETRIES_HEADER, new Integer(retries.intValue() + 1)) .setHeader(BinderHeaders.PARTITION_OVERRIDE, failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)) .build(); } else { System.out.println("Retries exhausted for " + failed); parkingLot.send(MessageBuilder.fromMessage(failed) .setHeader(BinderHeaders.PARTITION_OVERRIDE, failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)) .build()); } return null; } @Override public void run(String... args) throws Exception { while (true) { int count = this.processed.get(); Thread.sleep(5000); if (count == this.processed.get()) { System.out.println("Idle, terminating"); return; } } } public interface TwoOutputProcessor extends Processor { @Output("parkingLot") MessageChannel parkingLot(); } }
Apache Kafka supports topic partitioning natively.
Sometimes it is advantageous to send data to specific partitions — for example, when you want to strictly order message processing (all messages for a particular customer should go to the same partition).
The following example shows how to configure the producer and consumer side:
@SpringBootApplication @EnableBinding(Source.class) public class KafkaPartitionProducerApplication { private static final Random RANDOM = new Random(System.currentTimeMillis()); private static final String[] data = new String[] { "foo1", "bar1", "qux1", "foo2", "bar2", "qux2", "foo3", "bar3", "qux3", "foo4", "bar4", "qux4", }; public static void main(String[] args) { new SpringApplicationBuilder(KafkaPartitionProducerApplication.class) .web(false) .run(args); } @InboundChannelAdapter(channel = Source.OUTPUT, poller = @Poller(fixedRate = "5000")) public Message<?> generate() { String value = data[RANDOM.nextInt(data.length)]; System.out.println("Sending: " + value); return MessageBuilder.withPayload(value) .setHeader("partitionKey", value) .build(); } }
application.yml.
spring: cloud: stream: bindings: output: destination: partitioned.topic producer: partitioned: true partition-key-expression: headers['partitionKey'] partition-count: 12
![]() | Important |
---|---|
The topic must be provisioned to have enough partitions to achieve the desired concurrency for all consumer groups.
The above configuration supports up to 12 consumer instances (6 if their |
![]() | Note |
---|---|
The preceding configuration uses the default partitioning ( |
Since partitions are natively handled by Kafka, no special configuration is needed on the consumer side. Kafka allocates partitions across the instances.
The following Spring Boot application listens to a Kafka stream and prints (to the console) the partition ID to which each message goes:
@SpringBootApplication @EnableBinding(Sink.class) public class KafkaPartitionConsumerApplication { public static void main(String[] args) { new SpringApplicationBuilder(KafkaPartitionConsumerApplication.class) .web(false) .run(args); } @StreamListener(Sink.INPUT) public void listen(@Payload String in, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) { System.out.println(in + " received from partition " + partition); } }
application.yml.
spring: cloud: stream: bindings: input: destination: partitioned.topic group: myGroup
You can add instances as needed.
Kafka rebalances the partition allocations.
If the instance count (or instance count * concurrency
) exceeds the number of partitions, some consumers are idle.
For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId> </dependency>
Spring Cloud Stream’s Apache Kafka support also includes a binder implementation designed explicitly for Apache Kafka Streams binding. With this native integration, a Spring Cloud Stream "processor" application can directly use the Apache Kafka Streams APIs in the core business logic.
Kafka Streams binder implementation builds on the foundation provided by the Kafka Streams in Spring Kafka project.
Kafka Streams binder provides binding capabilities for the three major types in Kafka Streams - KStream, KTable and GlobalKTable.
As part of this native integration, the high-level Streams DSL provided by the Kafka Streams API is available for use in the business logic.
An early version of the Processor API support is available as well.
As noted early-on, Kafka Streams support in Spring Cloud Stream is strictly only available for use in the Processor model. A model in which the messages read from an inbound topic, business processing can be applied, and the transformed messages can be written to an outbound topic. It can also be used in Processor applications with a no-outbound destination.
This application consumes data from a Kafka topic (e.g., words
), computes word count for each unique word in a 5 seconds
time window, and the computed results are sent to a downstream topic (e.g., counts
) for further processing.
@SpringBootApplication @EnableBinding(KStreamProcessor.class) public class WordCountProcessorApplication { @StreamListener("input") @SendTo("output") public KStream<?, WordCount> process(KStream<?, String> input) { return input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .groupBy((key, value) -> value) .windowedBy(TimeWindows.of(5000)) .count(Materialized.as("WordCounts-multi")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); } public static void main(String[] args) { SpringApplication.run(WordCountProcessorApplication.class, args); }
Once built as a uber-jar (e.g., wordcount-processor.jar
), you can run the above example like the following.
java -jar wordcount-processor.jar --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts
This application will consume messages from the Kafka topic words
and the computed results are published to an output
topic counts
.
Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are automatically bound as KStream objects. As a developer, you can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor. Setting up the Streams DSL specific configuration required by the Kafka Streams infrastructure is automatically handled by the framework.
This section contains the configuration options used by the Kafka Streams binder.
For common configuration options and properties pertaining to binder, refer to the core documentation.
The following properties are available at the binder level and must be prefixed with spring.cloud.stream.kafka.streams.binder.
literal.
spring.cloud.stream.kafka.streams.binder.
.
Following are some examples of using this property.spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs in Apache Kafka Streams docs.
Broker URL
Default: localhost
Zookeeper URL
Default: localhost
Deserialization error handler type.
Possible values are - logAndContinue
, logAndFail
or sendToDlq
Default: logAndFail
Convenient way to set the application.id for the Kafka Streams application globally at the binder level.
If the application contains multiple StreamListener
methods, then application.id should be set at the binding level per input binding.
Default: none
The following properties are only available for Kafka Streams producers and must be prefixed with spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.
literal.
For convenience, if there multiple output bindings and they all require a common value, that can be configured by using the prefix spring.cloud.stream.kafka.streams.default.producer.
.
key serde to use
Default: none
.
value serde to use
Default: none
.
flag to enable native encoding
Default: false
.
The following properties are only available for Kafka Streams consumers and must be prefixed with spring.cloud.stream.kafka.streams.bindings.<binding name>.consumer.`literal.
For convenience, if there multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.
.
Setting application.id per input binding.
Default: none
key serde to use
Default: none
.
value serde to use
Default: none
.
state store to materialize when using incoming KTable types
Default: none
.
flag to enable native decoding
Default: false
.
DLQ topic name.
Default: none
.
Windowing is an important concept in stream processing applications. Following properties are available to configure time-window computations.
When this property is given, you can autowire a TimeWindows
bean into the application.
The value is expressed in milliseconds.
Default: none
.
Value is given in milliseconds.
Default: none
.
For use cases that requires multiple incoming KStream objects or a combination of KStream and KTable objects, the Kafka Streams binder provides multiple bindings support.
Let’s see it in action.
@EnableBinding(KStreamKTableBinding.class) ..... ..... @StreamListener public void process(@Input("inputStream") KStream<String, PlayEvent> playEvents, @Input("inputTable") KTable<Long, Song> songTable) { .... .... } interface KStreamKTableBinding { @Input("inputStream") KStream<?, ?> inputStream(); @Input("inputTable") KTable<?, ?> inputTable(); }
In the above example, the application is written as a sink, i.e. there are no output bindings and the application has to decide concerning downstream processing. When you write applications in this style, you might want to send the information downstream or store them in a state store (See below for Queryable State Stores).
In the case of incoming KTable, if you want to materialize the computations to a state store, you have to express it through the following property.
spring.cloud.stream.kafka.streams.bindings.inputTable.consumer.materializedAs: all-songs
The above example shows the use of KTable as an input binding. The binder also supports input bindings for GlobalKTable. GlobalKTable binding is useful when you have to ensure that all instances of your application has access to the data updates from the topic. KTable and GlobalKTable bindings are only available on the input. Binder supports both input and output bindings for KStream.
@EnableBinding(KStreamKTableBinding.class) .... .... @StreamListener @SendTo("output") public KStream<String, Long> process(@Input("input") KStream<String, Long> userClicksStream, @Input("inputTable") KTable<String, String> userRegionsTable) { .... .... } interface KStreamKTableBinding extends KafkaStreamsProcessor { @Input("inputX") KTable<?, ?> inputTable(); }
Kafka Streams allow outbound data to be split into multiple topics based on some predicates. The Kafka Streams binder provides
support for this feature without compromising the programming model exposed through StreamListener
in the end user application.
You can write the application in the usual way as demonstrated above in the word count example. However, when using the
branching feature, you are required to do a few things. First, you need to make sure that your return type is KStream[]
instead of a regular KStream
. Second, you need to use the SendTo
annotation containing the output bindings in the order
(see example below). For each of these output bindings, you need to configure destination, content-type etc., complying with
the standard Spring Cloud Stream expectations.
Here is an example:
@EnableBinding(KStreamProcessorWithBranches.class) @EnableAutoConfiguration public static class WordCountProcessorApplication { @Autowired private TimeWindows timeWindows; @StreamListener("input") @SendTo({"output1","output2","output3}) public KStream<?, WordCount>[] process(KStream<Object, String> input) { Predicate<Object, WordCount> isEnglish = (k, v) -> v.word.equals("english"); Predicate<Object, WordCount> isFrench = (k, v) -> v.word.equals("french"); Predicate<Object, WordCount> isSpanish = (k, v) -> v.word.equals("spanish"); return input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .groupBy((key, value) -> value) .windowedBy(timeWindows) .count(Materialized.as("WordCounts-1")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))) .branch(isEnglish, isFrench, isSpanish); } interface KStreamProcessorWithBranches { @Input("input") KStream<?, ?> input(); @Output("output1") KStream<?, ?> output1(); @Output("output2") KStream<?, ?> output2(); @Output("output3") KStream<?, ?> output3(); } }
Properties:
spring.cloud.stream.bindings.output1.contentType: application/json spring.cloud.stream.bindings.output2.contentType: application/json spring.cloud.stream.bindings.output3.contentType: application/json spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 1000 spring.cloud.stream.kafka.streams.binder.configuration: default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde spring.cloud.stream.bindings.output1: destination: foo producer: headerMode: raw spring.cloud.stream.bindings.output2: destination: bar producer: headerMode: raw spring.cloud.stream.bindings.output3: destination: fox producer: headerMode: raw spring.cloud.stream.bindings.input: destination: words consumer: headerMode: raw
Similar to message-channel based binder applications, the Kafka Streams binder adapts to the out-of-the-box content-type conversions without any compromise.
It is typical for Kafka Streams operations to know the type of SerDe’s used to transform the key and value correctly. Therefore, it may be more natural to rely on the SerDe facilities provided by the Apache Kafka Streams library itself at the inbound and outbound conversions rather than using the content-type conversions offered by the framework. On the other hand, you might be already familiar with the content-type conversion patterns provided by the framework, and that, you’d like to continue using for inbound and outbound conversions.
Both the options are supported in the Kafka Streams binder implementation.
If native encoding is disabled (which is the default), then the framework will convert the message using the contentType
set by the user (otherwise, the default application/json
will be applied). It will ignore any SerDe set on the outbound
in this case for outbound serialization.
Here is the property to set the contentType on the outbound.
spring.cloud.stream.bindings.output.contentType: application/json
Here is the property to enable native encoding.
spring.cloud.stream.bindings.output.nativeEncoding: true
If native encoding is enabled on the output binding (user has to enable it as above explicitly), then the framework will
skip any form of automatic message conversion on the outbound. In that case, it will switch to the Serde set by the user.
The valueSerde
property set on the actual output binding will be used. Here is an example.
spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
If this property is not set, then it will use the "default" SerDe: spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde
.
It is worth to mention that Kafka Streams binder does not serialize the keys on outbound - it simply relies on Kafka itself.
Therefore, you either have to specify the keySerde
property on the binding or it will default to the application-wide common
keySerde
.
Binding level key serde:
spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde
Common Key serde:
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde
If branching is used, then you need to use multiple output bindings. For example,
interface KStreamProcessorWithBranches { @Input("input") KStream<?, ?> input(); @Output("output1") KStream<?, ?> output1(); @Output("output2") KStream<?, ?> output2(); @Output("output3") KStream<?, ?> output3(); }
If nativeEncoding
is set, then you can set different SerDe’s on individual output bindings as below.
spring.cloud.stream.kafka.streams.bindings.output1.producer.valueSerde=IntegerSerde spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=StringSerde spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=JsonSerde
Then if you have SendTo
like this, @SendTo({"output1", "output2", "output3"}), the KStream[]
from the branches are
applied with proper SerDe objects as defined above. If you are not enabling nativeEncoding
, you can then set different
contentType values on the output bindings as below. In that case, the framework will use the appropriate message converter
to convert the messages before sending to Kafka.
spring.cloud.stream.bindings.output1.contentType: application/json spring.cloud.stream.bindings.output2.contentType: application/java-serialzied-object spring.cloud.stream.bindings.output3.contentType: application/octet-stream
Similar rules apply to data deserialization on the inbound.
If native decoding is disabled (which is the default), then the framework will convert the message using the contentType
set by the user (otherwise, the default application/json
will be applied). It will ignore any SerDe set on the inbound
in this case for inbound deserialization.
Here is the property to set the contentType on the inbound.
spring.cloud.stream.bindings.input.contentType: application/json
Here is the property to enable native decoding.
spring.cloud.stream.bindings.input.nativeDecoding: true
If native decoding is enabled on the input binding (user has to enable it as above explicitly), then the framework will
skip doing any message conversion on the inbound. In that case, it will switch to the SerDe set by the user. The valueSerde
property set on the actual output binding will be used. Here is an example.
spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
If this property is not set, it will use the default SerDe: spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde
.
It is worth to mention that Kafka Streams binder does not deserialize the keys on inbound - it simply relies on Kafka itself.
Therefore, you either have to specify the keySerde
property on the binding or it will default to the application-wide common
keySerde
.
Binding level key serde:
spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde
Common Key serde:
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde
As in the case of KStream branching on the outbound, the benefit of setting value SerDe per binding is that if you have multiple input bindings (multiple KStreams object) and they all require separate value SerDe’s, then you can configure them individually. If you use the common configuration approach, then this feature won’t be applicable.
Apache Kafka Streams provide the capability for natively handling exceptions from deserialization errors.
For details on this support, please see this
Out of the box, Apache Kafka Streams provide two kinds of deserialization exception handlers - logAndContinue
and logAndFail
.
As the name indicates, the former will log the error and continue processing the next records and the latter will log the
error and fail. LogAndFail
is the default deserialization exception handler.
Kafka Streams binder supports a selection of exception handlers through the following properties.
spring.cloud.stream.kafka.streams.binder.serdeError: logAndContinue
In addition to the above two deserialization exception handlers, the binder also provides a third one for sending the erroneous records (poison pills) to a DLQ topic. Here is how you enable this DLQ exception handler.
spring.cloud.stream.kafka.streams.binder.serdeError: sendToDlq
When the above property is set, all the deserialization error records are automatically sent to the DLQ topic.
spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: foo-dlq
If this is set, then the error records are sent to the topic foo-dlq
. If this is not set, then it will create a DLQ
topic with the name error.<input-topic-name>.<group-name>
.
A couple of things to keep in mind when using the exception handling feature in Kafka Streams binder.
spring.cloud.stream.kafka.streams.binder.serdeError
is applicable for the entire application. This implies
that if there are multiple StreamListener
methods in the same application, this property is applied to all of them.For general error handling in Kafka Streams binder, it is up to the end user applications to handle application level errors. As a side effect of providing a DLQ for deserialization exception handlers, Kafka Streams binder provides a way to get access to the DLQ sending bean directly from your application. Once you get access to that bean, you can programmatically send any exception records from your application to the DLQ.
It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn’t natively support error handling yet.
However, when you use the low-level Processor API in your application, there are options to control this behavior. See below.
@Autowired private SendToDlqAndContinue dlqHandler; @StreamListener("input") @SendTo("output") public KStream<?, WordCount> process(KStream<Object, String> input) { input.process(() -> new Processor() { ProcessorContext context; @Override public void init(ProcessorContext context) { this.context = context; } @Override public void process(Object o, Object o2) { try { ..... ..... } catch(Exception e) { //explicitly provide the kafka topic corresponding to the input binding as the first argument. //DLQ handler will correctly map to the dlq topic from the actual incoming destination. dlqHandler.sendToDlq("topic-name", (byte[]) o1, (byte[]) o2, context.partition()); } } ..... ..... }); }
State store is created automatically by Kafka Streams when the DSL is used.
When processor API is used, you need to register a state store manually. In order to do so, you can use KafkaStreamsStateStore
annotation.
You can specify the name and type of the store, flags to control log and disabling cache, etc.
Once the store is created by the binder during the bootstrapping phase, you can access this state store through the processor API.
Below are some primitives for doing this.
Creating a state store:
@KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs=300000) public void process(KStream<Object, Product> input) { ... }
Accessing the state store:
Processor<Object, Product>() { WindowStore<Object, String> state; @Override public void init(ProcessorContext processorContext) { state = (WindowStore)processorContext.getStateStore("mystate"); } ... }
As part of the public Kafka Streams binder API, we expose a class called InteractiveQueryService
.
You can access this as a Spring bean in your application. An easy way to get access to this bean from your application is to "autowire" the bean.
@Autowired private InteractiveQueryService interactiveQueryService;
Once you gain access to this bean, then you can query for the particular state-store that you are interested. See below.
ReadOnlyKeyValueStore<Object, Object> keyValueStore = interactiveQueryService.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
If there are multiple instances of the kafka streams application running, then before you can query them interactively, you need to identify which application instance hosts the key.
InteractiveQueryService
API provides methods for identifying the host information.
In order for this to work, you must configure the property application.server
as below:
spring.cloud.stream.kafka.streams.binder.configuration.application.server: <server>:<port>
Here are some code snippets:
org.apache.kafka.streams.state.HostInfo hostInfo = interactiveQueryService.getHostInfo("store-name", key, keySerializer); if (interactiveQueryService.getCurrentHostInfo().equals(hostInfo)) { //query from the store that is locally available } else { //query from the remote host }
StreamBuilderFactoryBean
from spring-kafka that is responsible for constructing the KafkaStreams
object can be accessed programmatically.
Each StreamBuilderFactoryBean
is registered as stream-builder
and appended with the StreamListener
method name.
If your StreamListener
method is named as process
for example, the stream builder bean is named as stream-builder-process
.
Since this is a factory bean, it should be accessed by prepending an ampersand (&
) when accessing it programmatically.
Following is an example and it assumes the StreamListener
method is named as process
StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
By default, the Kafkastreams.cleanup()
method is called when the binding is stopped.
See the Spring Kafka documentation.
To modify this behavior simply add a single CleanupConfig
@Bean
(configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean.
To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-stream-binder-rabbit</artifactId> </dependency>
Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>
The following simplified diagram shows how the RabbitMQ binder operates:
By default, the RabbitMQ Binder implementation maps each destination to a TopicExchange
.
For each consumer group, a Queue
is bound to that TopicExchange
.
Each consumer instance has a corresponding RabbitMQ Consumer
instance for its group’s Queue
.
For partitioned producers and consumers, the queues are suffixed with the partition index and use the partition index as the routing key.
For anonymous consumers (those with no group
property), an auto-delete queue (with a randomized unique name) is used.
By using the optional autoBindDlq
option, you can configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange DLX
, as well as routing infrastructure).
By default, the dead letter queue has the name of the destination, appended with .dlq
.
If retry is enabled (maxAttempts > 1
), failed messages are delivered to the DLQ after retries are exhausted.
If retry is disabled (maxAttempts = 1
), you should set requeueRejected
to false
(the default) so that failed messages are routed to the DLQ, instead of being re-queued.
In addition, republishToDlq
causes the binder to publish a failed message to the DLQ (instead of rejecting it).
This feature lets additional information (such as the stack trace in the x-exception-stacktrace
header) be added to the message in headers.
This option does not need retry enabled.
You can republish a failed message after just one attempt.
Starting with version 1.2, you can configure the delivery mode of republished messages.
See the republishDeliveryMode
property.
![]() | Important |
---|---|
Setting |
See Section 41.3.1, “RabbitMQ Binder Properties” for more information about these properties.
The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue). Some options are described in Section 41.6, “Dead-Letter Queue Processing”.
![]() | Note |
---|---|
When multiple RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from |
Starting with version 2.0, the RabbitMessageChannelBinder
sets the RabbitTemplate.userPublisherConnection
property to true
so that the non-transactional producers avoid deadlocks on consumers, which can happen if cached connections are blocked because of a memory alarm on the broker.
![]() | Note |
---|---|
Currently, a |
This section contains settings specific to the RabbitMQ Binder and bound channels.
For general binding configuration options and properties, see the Spring Cloud Stream core documentation.
By default, the RabbitMQ binder uses Spring Boot’s ConnectionFactory
.
Conseuqently, it supports all Spring Boot configuration options for RabbitMQ.
(For reference, see the Spring Boot documentation).
RabbitMQ configuration options use the spring.rabbitmq
prefix.
In addition to Spring Boot options, the RabbitMQ binder supports the following properties:
A comma-separated list of RabbitMQ management plugin URLs.
Only used when nodes
contains more than one entry.
Each entry in this list must have a corresponding entry in spring.rabbitmq.addresses
.
Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue.
See Queue Affinity and the LocalizedQueueConnectionFactory for more information.
Default: empty.
A comma-separated list of RabbitMQ node names.
When more than one entry, used to locate the server address where a queue is located.
Each entry in this list must have a corresponding entry in spring.rabbitmq.addresses
.
Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue.
See Queue Affinity and the LocalizedQueueConnectionFactory for more information.
Default: empty.
The compression level for compressed bindings.
See java.util.zip.Deflater
.
Default: 1
(BEST_LEVEL).
A connection name prefix used to name the connection(s) created by this binder.
The name is this prefix followed by #n
, where n
increments each time a new connection is opened.
Default: none (Spring AMQP default).
The following properties are available for Rabbit consumers only and must be prefixed with spring.cloud.stream.rabbit.bindings.<channelName>.consumer.
.
The acknowledge mode.
Default: AUTO
.
Whether to automatically declare the DLQ and bind it to the binder DLX.
Default: false
.
The routing key with which to bind the queue to the exchange (if bindQueue
is true
).
For partitioned destinations, -<instanceIndex>
is appended.
Default: #
.
Whether to bind the queue to the destination exchange.
Set it to false
if you have set up your own infrastructure and have previously created and bound the queue.
Default: true
.
Used to create the consumer tag(s); will be appended by #n
where n
increments for each consumer created.
Example: ${spring.application.name}-${spring.cloud.stream.bindings.input.group}-${spring.cloud.stream.instance-index}
.
Default: none - the broker will generate random consumer tags.
The name of the DLQ
Default: prefix+destination.dlq
A DLX to assign to the queue.
Relevant only if autoBindDlq
is true
.
Default: 'prefix+DLX'
The type of the DLX to assign to the queue.
Relevant only if autoBindDlq
is true
.
Default: 'direct'
A dead letter routing key to assign to the queue.
Relevant only if autoBindDlq
is true
.
Default: destination
Whether to declare the dead letter exchange for the destination.
Relevant only if autoBindDlq
is true
.
Set to false
if you have a pre-configured DLX.
Default: true
.
Whether to declare the exchange for the destination.
Default: true
.
Whether to declare the exchange as a Delayed Message Exchange
.
Requires the delayed message exchange plugin on the broker.
The x-delayed-type
argument is set to the exchangeType
.
Default: false
.
If a DLQ is declared, a DLX to assign to that queue.
Default: none
If a DLQ is declared, a dead letter routing key to assign to that queue.
Default: none
How long before an unused dead letter queue is deleted (in milliseconds).
Default: no expiration
Declare the dead letter queue with the x-queue-mode=lazy
argument.
See “Lazy Queues”.
Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue.
Default: false
.
Maximum number of messages in the dead letter queue.
Default: no limit
Maximum number of total bytes in the dead letter queue from all messages.
Default: no limit
Maximum priority of messages in the dead letter queue (0-255).
Default: none
Action to take when dlqMaxLength
or dlqMaxLengthBytes
is exceeded; currently drop-head
or reject-publish
but refer to the RabbitMQ documentation.
Default: none
Default time to live to apply to the dead letter queue when declared (in milliseconds).
Default: no limit
Whether the subscription should be durable.
Only effective if group
is also set.
Default: true
.
If declareExchange
is true, whether the exchange should be auto-deleted (that is, removed after the last queue is removed).
Default: true
.
If declareExchange
is true, whether the exchange should be durable (that is, it survives broker restart).
Default: true
.
The exchange type: direct
, fanout
or topic
for non-partitioned destinations and direct
or topic
for partitioned destinations.
Default: topic
.
Whether to create an exclusive consumer.
Concurrency should be 1 when this is true
.
Often used when strict ordering is required but enabling a hot standby instance to take over after a failure.
See recoveryInterval
, which controls how often a standby instance attempts to consume.
Default: false
.
How long before an unused queue is deleted (in milliseconds).
Default: no expiration
The interval (in milliseconds) between attempts to consume from a queue if it is missing.
Default: 5000
Patterns for headers to be mapped from inbound messages.
Default: ['*']
(all headers).
Declare the queue with the x-queue-mode=lazy
argument.
See “Lazy Queues”.
Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue.
Default: false
.
The maximum number of consumers.
Default: 1
.
The maximum number of messages in the queue.
Default: no limit
The maximum number of total bytes in the queue from all messages.
Default: no limit
The maximum priority of messages in the queue (0-255).
Default: none
When the queue cannot be found, whether to treat the condition as fatal and stop the listener container.
Defaults to false
so that the container keeps trying to consume from the queue — for example, when using a cluster and the node hosting a non-HA queue is down.
Default: false
Action to take when maxLength
or maxLengthBytes
is exceeded; currently drop-head
or reject-publish
but refer to the RabbitMQ documentation.
Default: none
Prefetch count.
Default: 1
.
A prefix to be added to the name of the destination
and queues.
Default: "".
The number of times to retry consuming from a queue if it is missing.
Relevant only when missingQueuesFatal
is true
.
Otherwise, the container keeps retrying indefinitely.
Default: 3
When true, consume from a queue with a name equal to the group
.
Otherwise the queue name is destination.group
.
This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue.
Default: false.
The interval between connection recovery attempts, in milliseconds.
Default: 5000
.
Whether delivery failures should be re-queued when retry is disabled or republishToDlq
is false
.
Default: false
.
When republishToDlq
is true
, specifies the delivery mode of the republished message.
Default: DeliveryMode.PERSISTENT
By default, messages that fail after retries are exhausted are rejected.
If a dead-letter queue (DLQ) is configured, RabbitMQ routes the failed message (unchanged) to the DLQ.
If set to true
, the binder republishs failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure.
Default: false
Whether to use transacted channels.
Default: false
.
Default time to live to apply to the queue when declared (in milliseconds).
Default: no limit
The number of deliveries between acks.
Default: 1
.
To set listener container properties that are not exposed as binder or binding properties, add a single bean of type ListenerContainerCustomizer
to the application context.
The binder and binding properties will be set and then the customizer will be called.
The customizer (configure()
method) is provided with the queue name as well as the consumer group as arguments.
The following properties are available for Rabbit producers only and
must be prefixed with spring.cloud.stream.rabbit.bindings.<channelName>.producer.
.
Whether to automatically declare the DLQ and bind it to the binder DLX.
Default: false
.
Whether to enable message batching by producers.
Messages are batched into one message according to the following properties (described in the next three entries in this list): 'batchSize', batchBufferLimit
, and batchTimeout
.
See Batching for more information.
Default: false
.
The number of messages to buffer when batching is enabled.
Default: 100
.
The maximum buffer size when batching is enabled.
Default: 10000
.
The batch timeout when batching is enabled.
Default: 5000
.
The routing key with which to bind the queue to the exchange (if bindQueue
is true
).
Only applies to non-partitioned destinations.
Only applies if requiredGroups
are provided and then only to those groups.
Default: #
.
Whether to bind the queue to the destination exchange.
Set it to false
if you have set up your own infrastructure and have previously created and bound the queue.
Only applies if requiredGroups
are provided and then only to those groups.
Default: true
.
Whether data should be compressed when sent.
Default: false
.
The name of the DLQ
Only applies if requiredGroups
are provided and then only to those groups.
Default: prefix+destination.dlq
A DLX to assign to the queue.
Relevant only when autoBindDlq
is true
.
Applies only when requiredGroups
are provided and then only to those groups.
Default: 'prefix+DLX'
The type of the DLX to assign to the queue.
Relevant only if autoBindDlq
is true
.
Applies only when requiredGroups
are provided and then only to those groups.
Default: 'direct'
A dead letter routing key to assign to the queue.
Relevant only when autoBindDlq
is true
.
Applies only when requiredGroups
are provided and then only to those groups.
Default: destination
Whether to declare the dead letter exchange for the destination.
Relevant only if autoBindDlq
is true
.
Set to false
if you have a pre-configured DLX.
Applies only when requiredGroups
are provided and then only to those groups.
Default: true
.
Whether to declare the exchange for the destination.
Default: true
.
A SpEL expression to evaluate the delay to apply to the message (x-delay
header).
It has no effect if the exchange is not a delayed message exchange.
Default: No x-delay
header is set.
Whether to declare the exchange as a Delayed Message Exchange
.
Requires the delayed message exchange plugin on the broker.
The x-delayed-type
argument is set to the exchangeType
.
Default: false
.
The delivery mode.
Default: PERSISTENT
.
When a DLQ is declared, a DLX to assign to that queue.
Applies only if requiredGroups
are provided and then only to those groups.
Default: none
When a DLQ is declared, a dead letter routing key to assign to that queue.
Applies only when requiredGroups
are provided and then only to those groups.
Default: none
How long (in milliseconds) before an unused dead letter queue is deleted.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no expiration
x-queue-mode=lazy
argument.
See “Lazy Queues”.
Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue.
Applies only when requiredGroups
are provided and then only to those groups.Maximum number of messages in the dead letter queue.
Applies only if requiredGroups
are provided and then only to those groups.
Default: no limit
Maximum number of total bytes in the dead letter queue from all messages.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no limit
Maximum priority of messages in the dead letter queue (0-255)
Applies only when requiredGroups
are provided and then only to those groups.
Default: none
Default time (in milliseconds) to live to apply to the dead letter queue when declared.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no limit
If declareExchange
is true
, whether the exchange should be auto-delete (it is removed after the last queue is removed).
Default: true
.
If declareExchange
is true
, whether the exchange should be durable (survives broker restart).
Default: true
.
The exchange type: direct
, fanout
or topic
for non-partitioned destinations and direct
or topic
for partitioned destinations.
Default: topic
.
How long (in milliseconds) before an unused queue is deleted.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no expiration
Patterns for headers to be mapped to outbound messages.
Default: ['*']
(all headers).
Declare the queue with the x-queue-mode=lazy
argument.
See “Lazy Queues”.
Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue.
Applies only when requiredGroups
are provided and then only to those groups.
Default: false
.
Maximum number of messages in the queue.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no limit
Maximum number of total bytes in the queue from all messages.
Only applies if requiredGroups
are provided and then only to those groups.
Default: no limit
Maximum priority of messages in the queue (0-255).
Only applies if requiredGroups
are provided and then only to those groups.
Default: none
A prefix to be added to the name of the destination
exchange.
Default: "".
When true
, consume from a queue with a name equal to the group
.
Otherwise the queue name is destination.group
.
This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue.
Applies only when requiredGroups
are provided and then only to those groups.
Default: false.
A SpEL expression to determine the routing key to use when publishing messages.
For a fixed routing key, use a literal expression, such as routingKeyExpression='my.routingKey'
in a properties file or routingKeyExpression: '''my.routingKey'''
in a YAML file.
Default: destination
or destination-<partition>
for partitioned destinations.
Whether to use transacted channels.
Default: false
.
Default time (in milliseconds) to live to apply to the queue when declared.
Applies only when requiredGroups
are provided and then only to those groups.
Default: no limit
![]() | Note |
---|---|
In the case of RabbitMQ, content type headers can be set by external applications. Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport — including transports, such as Kafka (prior to 0.11), that do not natively support headers. |
When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured. This might be important when strict ordering is required with a single consumer. However, for other use cases, it prevents other messages from being processed on that thread. An alternative to using binder retry is to set up dead lettering with time to live on the dead-letter queue (DLQ) as well as dead-letter configuration on the DLQ itself. See “Section 41.3.1, “RabbitMQ Binder Properties”” for more information about the properties discussed here. You can use the following example configuration to enable this feature:
autoBindDlq
to true
.
The binder create a DLQ.
Optionally, you can specify a name in deadLetterQueueName
.dlqTtl
to the back off time you want to wait between redeliveries.dlqDeadLetterExchange
to the default exchange.
Expired messages from the DLQ are routed to the original queue, because the default deadLetterRoutingKey
is the queue name (destination.group
).
Setting to the default exchange is achieved by setting the property with no value, as shown in the next example.To force a message to be dead-lettered, either throw an AmqpRejectAndDontRequeueException
or set requeueRejected
to true
(the default) and throw any exception.
The loop continue without end, which is fine for transient problems, but you may want to give up after some number of attempts.
Fortunately, RabbitMQ provides the x-death
header, which lets you determine how many cycles have occurred.
To acknowledge a message after giving up, throw an ImmediateAcknowledgeAmqpException
.
The following configuration creates an exchange myDestination
with queue myDestination.consumerGroup
bound to a topic exchange with a wildcard routing key #
:
--- spring.cloud.stream.bindings.input.destination=myDestination spring.cloud.stream.bindings.input.group=consumerGroup #disable binder retries spring.cloud.stream.bindings.input.consumer.max-attempts=1 #dlx/dlq setup spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=5000 spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange= ---
This configuration creates a DLQ bound to a direct exchange (DLX
) with a routing key of myDestination.consumerGroup
.
When messages are rejected, they are routed to the DLQ.
After 5 seconds, the message expires and is routed to the original queue by using the queue name as the routing key, as shown in the following example:
Spring Boot application.
@SpringBootApplication @EnableBinding(Sink.class) public class XDeathApplication { public static void main(String[] args) { SpringApplication.run(XDeathApplication.class, args); } @StreamListener(Sink.INPUT) public void listen(String in, @Header(name = "x-death", required = false) Map<?,?> death) { if (death != null && death.get("count").equals(3L)) { // giving up - don't send to DLX throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts"); } throw new AmqpRejectAndDontRequeueException("failed"); } }
Notice that the count property in the x-death
header is a Long
.
Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel. See “???” for more information.
RabbitMQ has two types of send failures:
The latter is rare. According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".
As well as enabling producer error channels (as described in “???”), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows:
ccf.setPublisherConfirms(true);
ccf.setPublisherReturns(true);
When using Spring Boot configuration for the connection factory, set the following properties:
spring.rabbitmq.publisher-confirms
spring.rabbitmq.publisher-returns
The payload of the ErrorMessage
for a returned message is a ReturnedAmqpMessageException
with the following properties:
failedMessage
: The spring-messaging Message<?>
that failed to be sent.amqpMessage
: The raw spring-amqp Message
.replyCode
: An integer value indicating the reason for the failure (for example, 312 - No route).replyText
: A text value indicating the reason for the failure (for example, NO_ROUTE
).exchange
: The exchange to which the message was published.routingKey
: The routing key used when the message was published.For negatively acknowledged confirmations, the payload is a NackedAmqpMessageException
with the following properties:
failedMessage
: The spring-messaging Message<?>
that failed to be sent.nackReason
: A reason (if available — you may need to examine the broker logs for more information).There is no automatic handling of these exceptions (such as sending to a dead-letter queue). You can consume these exceptions with your own Spring Integration flow.
Because you cannot anticipate how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them.
If the reason for the dead-lettering is transient, you may wish to route the messages back to the original queue.
However, if the problem is a permanent issue, that could cause an infinite loop.
The following Spring Boot application shows an example of how to route those messages back to the original queue but moves them to a third “parking lot” queue after three attempts.
The second example uses the RabbitMQ Delayed Message Exchange to introduce a delay to the re-queued message.
In this example, the delay increases for each attempt.
These examples use a @RabbitListener
to receive messages from the DLQ.
You could also use RabbitTemplate.receive()
in a batch process.
The examples assume the original destination is so8400in
and the consumer group is so8400
.
The first two examples are for when the destination is not partitioned:
@SpringBootApplication public class ReRouteDlqApplication { private static final String ORIGINAL_QUEUE = "so8400in.so8400"; private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; private static final String X_RETRIES_HEADER = "x-retries"; public static void main(String[] args) throws Exception { ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); System.out.println("Hit enter to terminate"); System.in.read(); context.close(); } @Autowired private RabbitTemplate rabbitTemplate; @RabbitListener(queues = DLQ) public void rePublish(Message failedMessage) { Integer retriesHeader = (Integer) failedMessage.getMessageProperties().getHeaders().get(X_RETRIES_HEADER); if (retriesHeader == null) { retriesHeader = Integer.valueOf(0); } if (retriesHeader < 3) { failedMessage.getMessageProperties().getHeaders().put(X_RETRIES_HEADER, retriesHeader + 1); this.rabbitTemplate.send(ORIGINAL_QUEUE, failedMessage); } else { this.rabbitTemplate.send(PARKING_LOT, failedMessage); } } @Bean public Queue parkingLot() { return new Queue(PARKING_LOT); } }
@SpringBootApplication public class ReRouteDlqApplication { private static final String ORIGINAL_QUEUE = "so8400in.so8400"; private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; private static final String X_RETRIES_HEADER = "x-retries"; private static final String DELAY_EXCHANGE = "dlqReRouter"; public static void main(String[] args) throws Exception { ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); System.out.println("Hit enter to terminate"); System.in.read(); context.close(); } @Autowired private RabbitTemplate rabbitTemplate; @RabbitListener(queues = DLQ) public void rePublish(Message failedMessage) { Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders(); Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); if (retriesHeader == null) { retriesHeader = Integer.valueOf(0); } if (retriesHeader < 3) { headers.put(X_RETRIES_HEADER, retriesHeader + 1); headers.put("x-delay", 5000 * retriesHeader); this.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage); } else { this.rabbitTemplate.send(PARKING_LOT, failedMessage); } } @Bean public DirectExchange delayExchange() { DirectExchange exchange = new DirectExchange(DELAY_EXCHANGE); exchange.setDelayed(true); return exchange; } @Bean public Binding bindOriginalToDelay() { return BindingBuilder.bind(new Queue(ORIGINAL_QUEUE)).to(delayExchange()).with(ORIGINAL_QUEUE); } @Bean public Queue parkingLot() { return new Queue(PARKING_LOT); } }
With partitioned destinations, there is one DLQ for all partitions. We determine the original queue from the headers.
When republishToDlq
is false
, RabbitMQ publishes the message to the DLX/DLQ with an x-death
header containing information about the original destination, as shown in the following example:
@SpringBootApplication public class ReRouteDlqApplication { private static final String ORIGINAL_QUEUE = "so8400in.so8400"; private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; private static final String X_DEATH_HEADER = "x-death"; private static final String X_RETRIES_HEADER = "x-retries"; public static void main(String[] args) throws Exception { ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); System.out.println("Hit enter to terminate"); System.in.read(); context.close(); } @Autowired private RabbitTemplate rabbitTemplate; @SuppressWarnings("unchecked") @RabbitListener(queues = DLQ) public void rePublish(Message failedMessage) { Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders(); Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); if (retriesHeader == null) { retriesHeader = Integer.valueOf(0); } if (retriesHeader < 3) { headers.put(X_RETRIES_HEADER, retriesHeader + 1); List<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get(X_DEATH_HEADER); String exchange = (String) xDeath.get(0).get("exchange"); List<String> routingKeys = (List<String>) xDeath.get(0).get("routing-keys"); this.rabbitTemplate.send(exchange, routingKeys.get(0), failedMessage); } else { this.rabbitTemplate.send(PARKING_LOT, failedMessage); } } @Bean public Queue parkingLot() { return new Queue(PARKING_LOT); } }
When republishToDlq
is true
, the republishing recoverer adds the original exchange and routing key to headers, as shown in the following example:
@SpringBootApplication public class ReRouteDlqApplication { private static final String ORIGINAL_QUEUE = "so8400in.so8400"; private static final String DLQ = ORIGINAL_QUEUE + ".dlq"; private static final String PARKING_LOT = ORIGINAL_QUEUE + ".parkingLot"; private static final String X_RETRIES_HEADER = "x-retries"; private static final String X_ORIGINAL_EXCHANGE_HEADER = RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE; private static final String X_ORIGINAL_ROUTING_KEY_HEADER = RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY; public static void main(String[] args) throws Exception { ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.class, args); System.out.println("Hit enter to terminate"); System.in.read(); context.close(); } @Autowired private RabbitTemplate rabbitTemplate; @RabbitListener(queues = DLQ) public void rePublish(Message failedMessage) { Map<String, Object> headers = failedMessage.getMessageProperties().getHeaders(); Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER); if (retriesHeader == null) { retriesHeader = Integer.valueOf(0); } if (retriesHeader < 3) { headers.put(X_RETRIES_HEADER, retriesHeader + 1); String exchange = (String) headers.get(X_ORIGINAL_EXCHANGE_HEADER); String originalRoutingKey = (String) headers.get(X_ORIGINAL_ROUTING_KEY_HEADER); this.rabbitTemplate.send(exchange, originalRoutingKey, failedMessage); } else { this.rabbitTemplate.send(PARKING_LOT, failedMessage); } } @Bean public Queue parkingLot() { return new Queue(PARKING_LOT); } }
RabbitMQ does not support partitioning natively.
Sometimes, it is advantageous to send data to specific partitions — for example, when you want to strictly order message processing, all messages for a particular customer should go to the same partition.
The RabbitMessageChannelBinder
provides partitioning by binding a queue for each partition to the destination exchange.
The following Java and YAML examples show how to configure the producer:
Producer.
@SpringBootApplication @EnableBinding(Source.class) public class RabbitPartitionProducerApplication { private static final Random RANDOM = new Random(System.currentTimeMillis()); private static final String[] data = new String[] { "abc1", "def1", "qux1", "abc2", "def2", "qux2", "abc3", "def3", "qux3", "abc4", "def4", "qux4", }; public static void main(String[] args) { new SpringApplicationBuilder(RabbitPartitionProducerApplication.class) .web(false) .run(args); } @InboundChannelAdapter(channel = Source.OUTPUT, poller = @Poller(fixedRate = "5000")) public Message<?> generate() { String value = data[RANDOM.nextInt(data.length)]; System.out.println("Sending: " + value); return MessageBuilder.withPayload(value) .setHeader("partitionKey", value) .build(); } }
application.yml.
spring: cloud: stream: bindings: output: destination: partitioned.destination producer: partitioned: true partition-key-expression: headers['partitionKey'] partition-count: 2 required-groups: - myGroup
![]() | Note |
---|---|
The configuration in the prececing example uses the default partitioning ( The |
The following configuration provisions a topic exchange:
The following queues are bound to that exchange:
The following bindings associate the queues to the exchange:
The following Java and YAML examples continue the previous examples and show how to configure the consumer:
Consumer.
@SpringBootApplication @EnableBinding(Sink.class) public class RabbitPartitionConsumerApplication { public static void main(String[] args) { new SpringApplicationBuilder(RabbitPartitionConsumerApplication.class) .web(false) .run(args); } @StreamListener(Sink.INPUT) public void listen(@Payload String in, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) { System.out.println(in + " received from queue " + queue); } }
application.yml.
spring: cloud: stream: bindings: input: destination: partitioned.destination group: myGroup consumer: partitioned: true instance-index: 0
![]() | Important |
---|---|
The |
Spring Cloud Bus links the nodes of a distributed system with a lightweight message broker. This broker can then be used to broadcast state changes (such as configuration changes) or other management instructions. A key idea is that the bus is like a distributed actuator for a Spring Boot application that is scaled out. However, it can also be used as a communication channel between apps. This project provides starters for either an AMQP broker or Kafka as the transport.
![]() | Note |
---|---|
Spring Cloud is released under the non-restrictive Apache 2.0 license. If you would like to contribute to this section of the documentation or if you find an error, please find the source code and issue trackers in the project at github. |
Spring Cloud Bus works by adding Spring Boot autconfiguration if it detects itself on the
classpath. To enable the bus, add spring-cloud-starter-bus-amqp
or
spring-cloud-starter-bus-kafka
to your dependency management. Spring Cloud takes care of
the rest. Make sure the broker (RabbitMQ or Kafka) is available and configured. When
running on localhost, you need not do anything. If you run remotely, use Spring Cloud
Connectors or Spring Boot conventions to define the broker credentials, as shown in the
following example for Rabbit:
application.yml.
spring: rabbitmq: host: mybroker.com port: 5672 username: user password: secret
The bus currently supports sending messages to all nodes listening or all nodes for a
particular service (as defined by Eureka). The /bus/*
actuator namespace has some HTTP
endpoints. Currently, two are implemented. The first, /bus/env
, sends key/value pairs to
update each node’s Spring Environment. The second, /bus/refresh
, reloads each
application’s configuration, as though they had all been pinged on their /refresh
endpoint.
![]() | Note |
---|---|
The Spring Cloud Bus starters cover Rabbit and Kafka, because those are the two most
common implementations. However, Spring Cloud Stream is quite flexible, and the binder
works with |
Spring Cloud Bus provides two endpoints, /actuator/bus-refresh
and /actuator/bus-env
that correspond to individual actuator endpoints in Spring Cloud Commons,
/actuator/refresh
and /actuator/env
respectively.
The /actuator/bus-refresh
endpoint clears the RefreshScope
cache and rebinds
@ConfigurationProperties
. See the Refresh Scope documentation for
more information.
To expose the /actuator/bus-refresh
endpoint, you need to add following configuration to your
application:
management.endpoints.web.exposure.include=bus-refresh
The /actuator/bus-env
endpoint updates each instances environment with the specified
key/value pair across multiple instances.
To expose the /actuator/bus-env
endpoint, you need to add following configuration to your
application:
management.endpoints.web.exposure.include=bus-env
The /actuator/bus-env
endpoint accepts POST
requests with the following shape:
{ "name": "key1", "value": "value1" }
Each instance of the application has a service ID, whose value can be set with
spring.cloud.bus.id
and whose value is expected to be a colon-separated list of
identifiers, in order from least specific to most specific. The default value is
constructed from the environment as a combination of the spring.application.name
and
server.port
(or spring.application.index
, if set). The default value of the ID is
constructed in the form of app:index:id
, where:
app
is the vcap.application.name
, if it exists, or spring.application.name
index
is the vcap.application.instance_index
, if it exists,
spring.application.index
, local.server.port
, server.port
, or 0
(in that order).id
is the vcap.application.instance_id
, if it exists, or a random value.The HTTP endpoints accept a “destination” path parameter, such as
/bus-refresh/customers:9000
, where destination
is a service ID. If the ID
is owned by an instance on the bus, it processes the message, and all other instances
ignore it.
The “destination” parameter is used in a Spring PathMatcher
(with the path separator
as a colon — :
) to determine if an instance processes the message. Using the example
from earlier, /bus-env/customers:**
targets all instances of the
“customers” service regardless of the rest of the service ID.
The bus tries twice to eliminate processing an event — once from the original
ApplicationEvent
and once from the queue. To do so, it checks the sending service ID
against the current service ID. If multiple instances of a service have the same ID,
events are not processed. When running on a local machine, each service is on a different
port, and that port is part of the ID. Cloud Foundry supplies an index to differentiate.
To ensure that the ID is unique outside Cloud Foundry, set spring.application.index
to
something unique for each instance of a service.
Spring Cloud Bus uses Spring Cloud Stream to
broadcast the messages. So, to get messages to flow, you need only include the binder
implementation of your choice in the classpath. There are convenient starters for the bus
with AMQP (RabbitMQ) and Kafka (spring-cloud-starter-bus-[amqp|kafka]
). Generally
speaking, Spring Cloud Stream relies on Spring Boot autoconfiguration conventions for
configuring middleware. For instance, the AMQP broker address can be changed with
spring.rabbitmq.*
configuration properties. Spring Cloud Bus has a handful of
native configuration properties in spring.cloud.bus.*
(for example,
spring.cloud.bus.destination
is the name of the topic to use as the external
middleware). Normally, the defaults suffice.
To learn more about how to customize the message broker settings, consult the Spring Cloud Stream documentation.
Bus events (subclasses of RemoteApplicationEvent
) can be traced by setting
spring.cloud.bus.trace.enabled=true
. If you do so, the Spring Boot TraceRepository
(if it is present) shows each event sent and all the acks from each service instance. The
following example comes from the /trace
endpoint:
{ "timestamp": "2015-11-26T10:24:44.411+0000", "info": { "signal": "spring.cloud.bus.ack", "type": "RefreshRemoteApplicationEvent", "id": "c4d374b7-58ea-4928-a312-31984def293b", "origin": "stores:8081", "destination": "*:**" } }, { "timestamp": "2015-11-26T10:24:41.864+0000", "info": { "signal": "spring.cloud.bus.sent", "type": "RefreshRemoteApplicationEvent", "id": "c4d374b7-58ea-4928-a312-31984def293b", "origin": "customers:9000", "destination": "*:**" } }, { "timestamp": "2015-11-26T10:24:41.862+0000", "info": { "signal": "spring.cloud.bus.ack", "type": "RefreshRemoteApplicationEvent", "id": "c4d374b7-58ea-4928-a312-31984def293b", "origin": "customers:9000", "destination": "*:**" } }
The preceding trace shows that a RefreshRemoteApplicationEvent
was sent from
customers:9000
, broadcast to all services, and received (acked) by customers:9000
and
stores:8081
.
To handle the ack signals yourself, you could add an @EventListener
for the
AckRemoteApplicationEvent
and SentApplicationEvent
types to your app (and enable
tracing). Alternatively, you could tap into the TraceRepository
and mine the data from
there.
![]() | Note |
---|---|
Any Bus application can trace acks. However, sometimes, it is useful to do this in a central service that can do more complex queries on the data or forward it to a specialized tracing service. |
The Bus can carry any event of type RemoteApplicationEvent
. The default transport is
JSON, and the deserializer needs to know which types are going to be used ahead of time.
To register a new type, you must put it in a subpackage of
org.springframework.cloud.bus.event
.
To customise the event name, you can use @JsonTypeName
on your custom class or rely on
the default strategy, which is to use the simple name of the class.
![]() | Note |
---|---|
Both the producer and the consumer need access to the class definition. |
If you cannot or do not want to use a subpackage of org.springframework.cloud.bus.event
for your custom events, you must specify which packages to scan for events of type
RemoteApplicationEvent
by using the @RemoteApplicationEventScan
annotation. Packages
specified with @RemoteApplicationEventScan
include subpackages.
For example, consider the following custom event, called MyEvent
:
package com.acme; public class MyEvent extends RemoteApplicationEvent { ... }
You can register that event with the deserializer in the following way:
package com.acme; @Configuration @RemoteApplicationEventScan public class BusConfiguration { ... }
Without specifying a value, the package of the class where @RemoteApplicationEventScan
is used is registered. In this example,