@SpringBootApplication
+@EnableBinding(Sink.class)
+public class SinkApplication {
+
+ private static Logger logger = LoggerFactory.getLogger(SinkModuleDefinition.class);
+
+ @ServiceActivator(inputChannel=Sink.INPUT)
+ public void loggerSink(Object payload) {
+ logger.info("Received: " + payload);
+ }
+}- Leader Election
@@ -789,6 +787,9 @@ 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.1.0.BUILD-SNAPSHOT
+@@ -1465,11 +1466,6 @@ list of properties), and "label" is an optional git label (defaults to "master".)--The YAML and properties forms are coalesced into a single -map, even if the origin of the values (reflected in the -"propertySources" of the "standard" form) has multiple sources.
-@@ -2010,6 +2006,20 @@ that should be applicable. HTTPS proxy settings can be set inSpring Cloud Config Server pulls configuration for remote clients from a git repository (which must be provided):
~/.git/configor in the same way as for any other JVM process via system properties (-Dhttps.proxyHostand-Dhttps.proxyPort). +++
++ ++ +Tip++If you don’t know where your +~/.gitdirectory is usgit config +--globalto manipulate the settings (e.g.git config --global +http.sslVerify false). +Placeholders in Git Search Paths
@@ -2529,6 +2539,51 @@ key nor the enabled flag.++Serving Alternative Formats
+++++The default JSON format from the environment endpoints is perfect for +consumption by Spring applications because it maps directly onto the +
+Environmentabstraction. If you prefer you can consume the same data +as YAML or Java properties by adding a suffix to the resource path +(".yml", ".yaml" or ".properties"). 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 fro mthe simplicity +of this approach.++The YAML and properties representations have an additional flag +(provided as a boolean query parameter
+resolvePlaceholders) to +signal that placeholders in the source documents, in the standard +Spring${…}form, should be resolved in the output where possible +before rendering. This is a useful feature for consumers that don’t +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. The JSON is structured as +an ordered list of property sources, for example, 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. 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. + ++Serving Plain Text
@@ -2760,23 +2815,25 @@ with the server address inspring.cloud.config.uri(defaults to-Eureka First Bootstrap
+Discovery First Bootstrap
-If you are using Spring Cloud Netflix and Eureka Service Discovery, -then you can have the Config Server register with Eureka if you want -to, but in the default "Config First" mode, clients won’t be able to -take advantage of the registration.
+If you are using a `DiscoveryClient implementation, such as Spring Cloud Netflix +and Eureka Service Discovery or Spring Cloud Consul (Spring Cloud Zookeeper does +not support this yet), then you can have the Config Server register with the +Discovery Service if you want to, but in the default "Config First" mode, +clients won’t be able to take advantage of the registration.
-@@ -4363,13 +4420,27 @@ independently for all the backends.If you prefer to use Eureka to locate the Config Server, you can do +
If you prefer to use
DiscoveryClientto locate the Config Server, you can do that by settingspring.cloud.config.discovery.enabled=true(default "false"). The net result of that is that client apps all need a -bootstrap.yml(or an environment variable) with the Eureka server -address, e.g. ineureka.client.serviceUrl.defaultZone. The price -for using this option is an extra network round trip on start up to +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, e.g. ineureka.client.serviceUrl.defaultZone. The +price for using this option is an extra network round trip on start up to locate the service registration. The benefit is that the Config Server -can change its co-ordinates, as long as Eureka is a fixed point. The -default service id is "CONFIGSERVER" but you can change that on the +can change its co-ordinates, as long as the Discovery Service is a fixed point. The +default service id is "configserver" but you can change that on the client withspring.cloud.config.discovery.serviceId(and on the server in the usual way for a service, e.g. by settingspring.application.name).+To enable it, annotate a Spring Boot main class with
@EnableZuulProxy, and this forwards local calls to the appropriate -service. By convention, a service with the Eureka ID "users", will +service. By convention, a service with the ID "users", will receive requests from the proxy located at/users(with the prefix stripped). The proxy uses Ribbon to locate an instance to forward to -via Eureka, and all requests are executed in a hystrix command, so +via discovery, and all requests are executed in a hystrix command, so failures will show up in Hystrix metrics, and once the circuit is open the proxy will 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 (e.g. Eureka is one choice). + +To skip having a service automatically added, set
zuul.ignored-servicesto a list of service id patterns. If a service @@ -5906,6 +5977,182 @@ Spring Cloud Stream supports them as part of an extended internal protocol used++Aggregation
+++Spring Cloud Stream provides support for aggregating multiple applications together, connecting their input and output channels directly and avoiding the additional cost of exchanging messages via a broker. +As of version 1.0 of Spring Cloud Stream, aggregation is supported only for the following types of applications:
+++-
+
-
+
sources - applications with a single output channel named
+output, typically having a single binding of the typeorg.springframework.cloud.stream.messaging.Source
+ -
+
sinks - applications with a single input channel named
+input, typically having a single binding of the typeorg.springframework.cloud.stream.messaging.Sink
+ -
+
processors - applications with a single input channel named
+inputand a single output channel namedoutput, typically having a single binding of the typeorg.springframework.cloud.stream.messaging.Processor.
+
++They can be aggregated together by creating a sequence of interconnected applications, in which the output channel of an element in the sequence is connected to the input channel of the next element, if it exists. +A sequence can start with either a source or a processor, it can contain an arbitrary number of processors and must end with either a processor or a sink.
+++Depending on the nature of the starting and ending element, the sequence may have one or more bindable channels, as follows:
+++-
+
-
+
if the sequence starts with a source and ends with a sink, all communication between the applications is direct and no channels will be bound
+
+ -
+
if the sequence starts with a processor, then its input channel will become the
+inputchannel of the aggregate and will be bound accordingly
+ -
+
if the sequence ends with a processor, then its output channel will become the
+outputchannel of the aggregate and will be bound accordingly
+
++Aggregation is performed using the
+AggregateApplicationBuilderutility class, as in the following example. +Let’s consider a project in which we have source, processor and a sink, which may be defined in the project, or may be contained in one of the project’s dependencies.++++++++
+@SpringBootApplication +@EnableBinding(Processor.class) +public class ProcessorApplication { + + @Transformer + public String loggerSink(String payload) { + return payload.toUpperCase(); + } +}++++
+@SpringBootApplication +@EnableBinding(Source.class) +public class SourceApplication { + + @Bean + @InboundChannelAdapter(value = Source.OUTPUT) + public String timerMessageSource() { + return new SimpleDateFormat().format(new Date()); + } +}++Each configuration can be used for running a separate component, but in this case they can be aggregated together as follows:
+++++
+@SpringBootApplication +public class SampleAggregateApplication { + + public static void main(String[] args) { + new AggregateApplicationBuilder() + .from(SourceApplication.class).args("--fixedDelay=5000") + .via(ProcessorApplication.class) + .to(SinkApplication.class).args("--debug=true").run(args); + } +}++The starting component of the sequence is provided as argument to the
+from()method. +The ending component of the sequence is provided as argument to theto()method. +Intermediate processors are provided as argument to thevia()method. +Multiple processors of the same type can be chained together (e.g. for pipelining transformations with different configurations). +For each component, the builder can provide runtime arguments for Spring Boot configuration.+RxJava support
+++Spring Cloud Stream provides support for RxJava-based processors through the
+RxJavaProcessoravailable inspring-cloud-stream-rxjava.++++
+public interface RxJavaProcessor<I, O> { + Observable<O> process(Observable<I> input); +}++An implementation of
+RxJavaProcessorwill receiveObservableas an input that represents the flow of inbound message payloads. +Theprocessmethod is invoked once at startup for setting up the data flow.++You can enable the use of RxJava-based processors and use them in your processor application by using the
+@EnableRxJavaProcessorannotation. +@EnableRxJavaProcessoris meta-annotated with@EnableBinding(Processor.class)and will create theProcessorbinding. +Here is an example of an RxJava-based processor:++++
+@EnableRxJavaProcessor +public class RxJavaTransformer { + + private static Logger logger = LoggerFactory.getLogger(RxJavaTransformer.class); + + @Bean + public RxJavaProcessor<String,String> processor() { + return inputStream -> inputStream.map(data -> { + logger.info("Got data = " + data); + return data; + }) + .buffer(5) + .map(data -> String.valueOf(avg(data))); + } + + private static Double avg(List<String> data) { + double sum = 0; + double count = 0; + for(String d : data) { + count++; + sum += Double.valueOf(d); + } + return sum/count; + } +}+++
++ ++ +Note++ +++When implementing an RxJava processor, it is important to handle exceptions as part of your processing flow. +Uncaught exceptions will be treated as errors by RxJava and will cause the
+Observableto complete, disrupting the flow.@@ -6513,11 +6797,13 @@ Allowed values:Binder SPI
@@ -6478,6 +6725,43 @@ Mutually exclusive withoffsetUpdateTimeWindow.- spring.cloud.stream.kafka.binder.requiredAcks
The number of required acks on the broker.
+++Default:
+1.- spring.cloud.stream.kafka.binder.minPartitionCount
+- +
+Effective only if
+autoCreateTopicsorautoAddPartitionsis set. + The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data. +It can be superseded by thepartitionCountsetting of the producer or by the value of +instanceCount*concurrencysettings of the producer (if either is larger).++Default:
+1.- spring.cloud.stream.kafka.binder.replicationFactor
+- +
+The replication factor of auto-created topics if
+autoCreateTopicsis active.++Default:
+1.- spring.cloud.stream.kafka.binder.autoCreateTopics
+- +
+If set to
+true, the binder will create new topics automatically. +If set tofalse, the binder will rely on the topics being already configured. +In the latter case, if the topics do not exist, the binder will fail to start. +Of note, this setting is independent of theauto.topic.create.enablesetting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings.++Default:
+true.- spring.cloud.stream.kafka.binder.autoAddPartitions
+- +
If set to
true, the binder will create add new partitions if required. +If set tofalse, the binder will rely on the partition size of the topic being already configured. +If the partition count of the target topic is smaller than the expected value, the binder will fail to start. +Default:false.earliest,latest.Default: null (equivalent to
-earliest).- minPartitionCount
+- enableDlq
- -
@@ -6724,7 +7010,7 @@ Spring Cloud Stream does this through theThe minimum number of partitions expected by the consumer if it creates the consumed topic automatically.
+When set to true, it will send enable DLQ behavior for the consumer. +Messages that result in errors will be forwarded to a topic named
error.<destination>.<group>. +This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.-Default:
+1.Default:
false.spring.cloud.stream.instanceCoun For example, if there are three instances of a HDFS sink application, all three instances will havespring.cloud.stream.instanceCountset to3, and the individual applications will havespring.cloud.stream.instanceIndexset to0,1, and2, respectively.-When Spring Cloud Stream applications are deployed via Spring Cloud Dataflow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. +
When Spring Cloud Stream applications are deployed via Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly. By default,
spring.cloud.stream.instanceCountis1, andspring.cloud.stream.instanceIndexis0.@@ -6741,8 +7027,8 @@ For example, the following is a valid and typical configuration:-spring.cloud.stream.bindings.output.partitionKeyExpression=payload.id -spring.cloud.stream.bindings.output.partitionCount=5
+spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload.id +spring.cloud.stream.bindings.output.producer.partitionCount=5
@@ -6775,6 +7061,24 @@ This can be customized on the binding, either by setting a SpEL expression to be-+Additional properties can be configured for more advanced scenarios, as described in the following section.
++
++ ++ +Note++ +++The Kafka binder will use the
+partitionCountsetting as a hint to create a topic with the given partition count (in conjunction with theminPartitionCount, the maximum of the two being the value being used). +Exercise caution when configuring bothminPartitionCountfor a binder andpartitionCountfor an application, as the larger value will be used. +If a topic already exists with a smaller partition count andautoAddPartitionsis disabled (the default), then the binder will fail to start. +If a topic already exists with a smaller partition count andautoAddPartitionsis enabled, new partitions will be added. +If a topic already exists with a larger number of partitions than the maximum of (minPartitionCountandpartitionCount), the existing partition count will be used.@@ -7341,9 +7657,11 @@ to Zipkin and merged there.Configuring Input Bindings for Partitioning
@@ -6782,7 +7086,7 @@ This can be customized on the binding, either by setting a SpEL expression to be-@@ -7025,7 +7329,7 @@ Spring Cloud Bus links nodes of a distributed system with a lightweight messagespring.cloud.stream.bindings.input.partitioned=true +
spring.cloud.stream.bindings.input.consumer.partitioned=true spring.cloud.stream.instanceIndex=3 spring.cloud.stream.instanceCount=5
Quick Start
-Spring Cloud Bus works by adding Spring Boot autconfiguration if it detects itself on the classpath. All you need to do to enable the bus is to add
+spring-cloud-starter-bus-amqpto your dependency management and Spring Cloud takes care of the rest. Make sure RabbitMQ is available and configured to provide aConnectionFactory: running on localhost you shouldn’t have to do anything, but if you are running remotely use Spring Cloud Connectors, or Spring Boot conventions to define the broker credentials, e.g.Spring Cloud Bus works by adding Spring Boot autconfiguration if it detects itself on the classpath. All you need to do to enable the bus is to add
spring-cloud-starter-bus-amqporspring-cloud-starter-bus-kafkato your dependency management and Spring Cloud takes care of the rest. Make sure the broker (RabbitMQ or Kafka) is available and configured: running on localhost you shouldn’t have to do anything, but if you are running remotely use Spring Cloud Connectors, or Spring Boot conventions to define the broker credentials, e.g. for Rabbitapplication.yml@@ -7039,7 +7343,19 @@ Spring Cloud Bus links nodes of a distributed system with a lightweight message-+The bus currently supports sending messages to all nodes listening or all nodes for a particular service (as defined by Eureka). More selector criteria will be added in the future (ie. only service X nodes in data center Y, etc…). The http endpoints are under the
+/bus/*actuator namespace. There are currently two implemented. The first,/bus/env, sends key/values pairs to update each nodes Spring Environment. The second,/bus/refresh, will reload each application’s configuration, just as if they had all been pinged on their/refreshendpoint.The bus currently supports sending messages to all nodes listening or all nodes for a particular service (as defined by Eureka). More selector criteria may be added in the future (ie. only service X nodes in data center Y, etc…). There are also some http endpoints are under the
+/bus/*actuator namespace. There are currently two implemented. The first,/bus/env, sends key/values pairs to update each nodes Spring Environment. The second,/bus/refresh, will reload each application’s configuration, just as if they had all been pinged on their/refreshendpoint.++
+ ++ +Note++the Bus starters cover Rabbit and Kafka, because those are the two most common implementations, but Spring Cloud Stream is quite flexible and binder will work combined with +spring-cloud-bus. +So 1 span from A, 2 spans from B, 1 span from C, 2 spans from D, 1 span from E, 2 spans from F and 1 from G. Altogether 10 spans.
--Click Pivotal Web Services icon to see it live!-
+
+++
+Click Pivotal Web Services icon to see it live!Click Pivotal Web Services icon to see it live!The dependency graph in Zipkin would look like this:
@@ -7353,9 +7671,11 @@ Altogether 10 spans.
-Click Pivotal Web Services icon to see it live!-
++++
+Click Pivotal Web Services icon to see it live!Click Pivotal Web Services icon to see it live!@@ -8359,8 +8679,8 @@ however will be still there.@Async annotated methods
-In Spring Cloud Sleuth we’re instrumenting async related components so that the tracing information is passed between threads. You can disable this behaviour -by setting the value of
+spring.sleuth.async.enabledtofalse.In Spring Cloud Sleuth we’re instrumenting async related components so that the tracing information is passed between threads. +You can disable this behaviour by setting the value of
spring.sleuth.async.enabledtofalse.If you annotate your method with
@@ -9817,8 +10137,8 @@ implementation of Spring Cloud Commons@Asyncthen we’ll automatically create a new Span with the following characteristics:DiscoveryClientso you can@EnableDiscoveryClientand provide your credentials asspring.cloud.cloudfoundry.discovery.[email,password]and then you can use theDiscoveryClientdirectly or via aLoadBalancerClient-(also*.urlif you are not connecting to [Pivotal Web -Services](https://run.pivotal.io)). +(also*.urlif you are not connecting to +Pivotal Web Services).The first time you use it the discovery client might be slow owing to @@ -9827,7 +10147,7 @@ the fact that it has to get an access token from Cloud Foundry.
-Quickstart
+Discovery
+Here’s a Spring Cloud app with Cloud Foundry discovery:
@@ -9863,8 +10183,17 @@ $ cf push -p app.jar-It will show its app name in the home page.
-+Single Sign On
+++The
+DiscoveryClientcan lists all the apps in a space, according to +the credentials it is authenticated with, where the space defaults to +the one the client is running in (if any). If neither org nor space +are configured, they default per the user’s profile in Cloud Foundry.+Single Sign On
+@@ -9890,7 +10219,6 @@ parameterized using spring.oauth2.sso.serviceId. -Spring Cloud Cluster
@@ -12024,7 +12352,7 @@ created during auto-configuration. -
+