diff --git a/docs/src/main/asciidoc/rabbit/images/part-bindings.png b/docs/modules/ROOT/assets/images/part-bindings.png similarity index 100% rename from docs/src/main/asciidoc/rabbit/images/part-bindings.png rename to docs/modules/ROOT/assets/images/part-bindings.png diff --git a/docs/src/main/asciidoc/rabbit/images/part-exchange.png b/docs/modules/ROOT/assets/images/part-exchange.png similarity index 100% rename from docs/src/main/asciidoc/rabbit/images/part-exchange.png rename to docs/modules/ROOT/assets/images/part-exchange.png diff --git a/docs/src/main/asciidoc/rabbit/images/part-queues.png b/docs/modules/ROOT/assets/images/part-queues.png similarity index 100% rename from docs/src/main/asciidoc/rabbit/images/part-queues.png rename to docs/modules/ROOT/assets/images/part-queues.png diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index 570cbbcf8..c203fb228 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -34,7 +34,7 @@ * xref:preface.adoc[] * xref:binders.adoc[] * xref:README.adoc[] -* xref:_configprops.adoc[] +* xref:configprops.adoc[] * xref:intro.adoc[] ** xref:kafka/kafka-reactive.adoc[] ** xref:kafka/kafka-streams.adoc[] diff --git a/docs/src/main/asciidoc/README.adoc b/docs/modules/ROOT/pages/README.adoc similarity index 66% rename from docs/src/main/asciidoc/README.adoc rename to docs/modules/ROOT/pages/README.adoc index e648bdb08..7ee7509cc 100644 --- a/docs/src/main/asciidoc/README.adoc +++ b/docs/modules/ROOT/pages/README.adoc @@ -11,20 +11,10 @@ image::https://codecov.io/gh/spring-cloud/spring-cloud-stream/branch/{github-tag // ====================================================================================== [[introduction]] -= Introduction +== Introduction [[resources]] -= Resources +== Resources -For more information, please visit the https://spring.io/projects/spring-cloud-stream[project website]: - -[[building]] -= Building - -include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/main/docs/modules/ROOT/partials/contributing.adoc[] - -[[contributing]] -= Contributing - -include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/docs/src/main/asciidoc/contributing.adoc[] +For more information, please visit the https://spring.io/projects/spring-cloud-stream[project website]. diff --git a/docs/modules/ROOT/pages/binders.adoc b/docs/modules/ROOT/pages/binders.adoc index a70a048e5..0ceb4d110 100644 --- a/docs/modules/ROOT/pages/binders.adoc +++ b/docs/modules/ROOT/pages/binders.adoc @@ -1,4 +1,3 @@ -*{project-version}* [[binders]] == Binder Implementations diff --git a/docs/modules/ROOT/pages/configprops.adoc b/docs/modules/ROOT/pages/configprops.adoc index 5187c48df..ec87f4391 100644 --- a/docs/modules/ROOT/pages/configprops.adoc +++ b/docs/modules/ROOT/pages/configprops.adoc @@ -3,4 +3,4 @@ Below you can find a list of configuration properties. -include::configprops.adoc[] + diff --git a/docs/modules/ROOT/pages/index.adoc b/docs/modules/ROOT/pages/index.adoc index 8cc88a560..4dc0d71c7 100644 --- a/docs/modules/ROOT/pages/index.adoc +++ b/docs/modules/ROOT/pages/index.adoc @@ -2,8 +2,6 @@ = Spring Cloud Stream Reference Documentation Sabby Anandan; Marius Bogoevici; Eric Bottard; Mark Fisher; Ilayaperumal Gopinathan; Mark Heckler; Gunnar Hillert; Mark Pollack; Patrick Peralta; Glenn Renfro; Thomas Risberg; Dave Syer; David Turanski; Janne Valkealahti; Benjamin Klein; Vinicius Carvalho; Gary Russell; Oleg Zhurakousky; Jay Bryant; Soby Chacko; Domenico Sibilio -*{project-version}* - :docinfo: shared The reference documentation consists of the following sections: @@ -12,8 +10,8 @@ The reference documentation consists of the following sections: xref:spring-cloud-stream.adoc#spring-cloud-stream-reference[Overview] :: History, Quick Start, Concepts, Architecture Overview, Binder Abstraction, and Core Features xref:rabbit/spring-cloud-stream-binder-rabbit.adoc[Rabbit MQ Binder] :: Spring Cloud Stream binder reference for Rabbit MQ -<> :: Spring Cloud Stream binder reference for Apache Kafka -<> :: Spring Cloud Stream binder reference for Apache Kafka Streams +xref:kafka/spring-cloud-stream-binder-kafka.adoc[Apache Kafka Binder] :: Spring Cloud Stream binder reference for Apache Kafka +xref:kafka/kafka-streams.adoc[Apache Kafka Streams Binder] :: Spring Cloud Stream binder reference for Apache Kafka Streams xref:pulsar/spring-cloud-stream-binder-pulsar.adoc[Apache Pulsar Binder] :: Spring Cloud Stream binder reference for Apache Pulsar https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#spring-cloud-stream-binder-for-solace-pubsub[Solace PubSub+ Binder] :: Spring Cloud Stream binder reference for Solace PubSub+ (Partner Maintained) diff --git a/docs/modules/ROOT/pages/intro.adoc b/docs/modules/ROOT/pages/intro.adoc index 5c0d10a16..f109776e3 100644 --- a/docs/modules/ROOT/pages/intro.adoc +++ b/docs/modules/ROOT/pages/intro.adoc @@ -8,10 +8,10 @@ While these two binder implementations are based on Message Channels, Spring Clo Below, you can find more information on how to use these various out-of-the-box binder implementations in your applications. [[apache-kafka-binder]] -= Apache Kafka Binder +== Apache Kafka Binder [[usage]] -== Usage +=== Usage 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: @@ -34,10 +34,10 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown ---- [[apache-kafka-streams-binder]] -= Apache Kafka Streams Binder +== Apache Kafka Streams Binder -[[usage]] -== Usage +[[apache-kafka-streams-binder-usage]] +=== Usage To use Apache Kafka Streams binder, you need to add `spring-cloud-stream-binder-kafka-streams` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven: @@ -50,9 +50,9 @@ To use Apache Kafka Streams binder, you need to add `spring-cloud-stream-binder- ---- [[rabbitmq-binder]] -= RabbitMQ Binder -[[usage]] -== Usage +== RabbitMQ Binder +[[rabbitmq-binder-usage]] +=== Usage To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates: diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams.adoc index 6d289091a..3cc1b098c 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams.adoc @@ -1,7 +1,7 @@ -== Kafka Streams Binder += Kafka Streams Binder -=== Usage +== Usage For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following maven coordinates: @@ -15,9 +15,9 @@ For using the Kafka Streams binder, you just need to add it to your Spring Cloud A quick way to bootstrap a new project for Kafka Streams binder is to use http://start.spring.io[Spring Initializr] and then select "Cloud Streams" and "Spring for Kafka Streams" as shown below -image::{github-raw}/docs/src/main/asciidoc/images/spring-initializr-kafka-streams.png[width=800,scaledwidth="75%",align="center"] +// image::{github-raw}/docs/src/main/asciidoc/images/spring-initializr-kafka-streams.png[width=800,scaledwidth="75%",align="center"] -=== Overview +== Overview Spring Cloud Stream includes a binder implementation designed explicitly for https://kafka.apache.org/documentation/streams/[Apache Kafka Streams] binding. With this native integration, a Spring Cloud Stream "processor" application can directly use the @@ -32,12 +32,12 @@ Alternatively, a Processor application with no outbound destination can be defin In the following sections, we are going to look at the details of Spring Cloud Stream's integration with Kafka Streams. -=== Programming Model +== Programming Model When using the programming model provided by Kafka Streams binder, both the high-level https://docs.confluent.io/current/streams/developer-guide/dsl-api.html[Streams DSL] and a mix of both the higher level and the lower level https://docs.confluent.io/current/streams/developer-guide/processor-api.html[Processor-API] can be used as options. When mixing both higher and lower level API's, this is usually achieved by invoking `transform` or `process` API methods on `KStream`. -==== Functional Style +=== Functional Style Starting with Spring Cloud Stream `3.0.0`, Kafka Streams binder allows the applications to be designed and developed using the functional programming style that is available in Java 8. This means that the applications can be concisely represented as a lambda expression of types `java.util.function.Function` or `java.util.function.Consumer`. @@ -155,7 +155,7 @@ In addition, you can also provide topic patterns as destinations if you want to `spring.cloud.stream.bindings.process-in-0.destination=input.*` -===== Multiple Input Bindings +=== Multiple Input Bindings Many non-trivial Kafka Streams applications often consume data from more than one topic through multiple bindings. For instance, one topic is consumed as `Kstream` and another as `KTable` or `GlobalKTable`. @@ -164,7 +164,7 @@ Think of a use-case where the underlying topic is populated through a change dat If the application specifies that the data needs to be bound as `KTable` or `GlobalKTable`, then Kafka Streams binder will properly bind the destination to a `KTable` or `GlobalKTable` and make them available for the application to operate upon. We will look at a few different scenarios how multiple input bindings are handled in the Kafka Streams binder. -====== BiFunction in Kafka Streams Binder +=== BiFunction in Kafka Streams Binder Here is an example where we have two inputs and an output. In this case, the application can leverage on `java.util.function.BiFunction`. @@ -189,7 +189,7 @@ Java's `BiFunction` support is used to bind the inputs to the desired destinatio The default binding names generated by the binder for the inputs are `process-in-0` and `process-in-1` respectively. The default output binding is `process-out-0`. In this example, the first parameter of `BiFunction` is bound as a `KStream` for the first input and the second parameter is bound as a `KTable` for the second input. -====== BiConsumer in Kafka Streams Binder +=== BiConsumer in Kafka Streams Binder If there are two inputs, but no outputs, in that case we can use `java.util.function.BiConsumer` as shown below. @@ -201,7 +201,7 @@ public BiConsumer, KTable> process() { } ---- -====== Beyond two inputs +=== Beyond two inputs What if you have more than two inputs? There are situations in which you need more than two inputs. In that case, the binder allows you to chain partial functions. @@ -254,7 +254,7 @@ Input bindings are named as `enrichOrder-in-0`, `enrichOrder-in-1` and `enrichOr With curried functions, you can virtually have any number of inputs. However, keep in mind that, anything more than a smaller number of inputs and partially applied functions for them as above in Java might lead to unreadable code. Therefore if your Kafka Streams application requires more than a reasonably smaller number of input bindings, and you want to use this functional model, then you may want to rethink your design and decompose the application appropriately. -===== Output Bindings +=== Output Bindings Kafka Streams binder allows types of either `KStream` or `KTable` as output bindings. Behind the scenes, the binder uses the `to` method on `KStream` to send the resultant records to the output topic. @@ -275,7 +275,7 @@ public Function, KStream> bar() { } ``` -===== Multiple Output Bindings +=== Multiple Output Bindings Kafka Streams allows writing outbound data into multiple topics. This feature is known as branching in Kafka Streams. When using multiple output bindings, you need to provide an array of KStream (`KStream[]`) as the outbound return type. @@ -316,7 +316,7 @@ The default output binding names are `process-out-0`, `process-out-1`, `process- The reason why the binder generates three output bindings is because it detects the length of the returned `KStream` array as three. Note that in this example, we provide a `noDefaultBranch()`; if we have used `defaultBranch()` instead, that would have required an extra output binding, essentially returning a `KStream` array of length four. -===== Summary of Function based Programming Styles for Kafka Streams +=== Summary of Function based Programming Styles for Kafka Streams In summary, the following table shows the various options that can be used in the functional paradigm. @@ -333,7 +333,7 @@ In summary, the following table shows the various options that can be used in th * In the case of more than one output in this table, the type simply becomes `KStream[]`. -===== Function composition in Kafka Streams binder +== Function composition in Kafka Streams binder Kafka Streams binder supports minimal forms of functional composition for linear topologies. Using the Java functional API support, you can write multiple functions and then compose them on your own using the `andThen` method. @@ -377,7 +377,7 @@ spring.cloud.function.definition=foo|bar;foo;bar The composed function's default binding names in this example becomes `foobar-in-0` and `foobar-out-0`. -====== Limitations of functional composition in Kafka Streams bincer +=== Limitations of functional composition in Kafka Streams bincer When you have `java.util.function.Function` bean, that can be composed with another function or multiple functions. The same function bean can be composed with a `java.util.function.Consumer` as well. In this case, consumer is the last component composed. @@ -411,7 +411,7 @@ Behind the scenes, the binder will create two input bindings for the curried fun The default input bindings in this case are going to be `curriedFoobar-in-0` and `curriedFoobar-in-1`. The default output binding for this example becomes `curriedFoobar-out-0`. -====== Special note on using `KTable` as output in function composition +=== Special note on using `KTable` as output in function composition Lets say you have the following two functions. @@ -430,9 +430,9 @@ public Function, KStream> bar() { You can compose them as `foo|bar`, but keep in mind that the second function (`bar` in this case) must have a `KTable` as input since the first function (`foo`) has `KTable` as output. -=== Ancillaries to the programming model +== Ancillaries to the programming model -==== Multiple Kafka Streams processors within a single application +=== Multiple Kafka Streams processors within a single application Binder allows to have multiple Kafka Streams processors within a single Spring Cloud Stream application. You can have an application as below. @@ -465,7 +465,7 @@ If you want certain functions to be not activated right away, you can remove tha This is also true when you have a single Kafka Streams processor and other types of `Function` beans in the same application that is handled through a different binder (for e.g., a function bean that is based on the regular Kafka Message Channel binder) -==== Kafka Streams Application ID +== Kafka Streams Application ID Application id is a mandatory property that you need to provide for a Kafka Streams application. Spring Cloud Stream Kafka Streams binder allows you to configure this application id in multiple ways. @@ -516,13 +516,13 @@ This is convenient in development scenarios as it avoids the need for explicitly The generated application ID in this manner will be static over application restarts. In the case of functional model, the generated application ID will be the function bean name followed by the literal `applicationID`, for e.g `process-applicationID` if `process` if the function bean name. -====== Summary of setting Application ID +=== Summary of setting Application ID * By default, binder will auto generate the application ID per function methods. * If you have a single processor, then you can use `spring.kafka.streams.applicationId`, `spring.application.name` or `spring.cloud.stream.kafka.streams.binder.applicationId`. * If you have multiple processors, then application ID can be set per function using the property - `spring.cloud.stream.kafka.streams.binder.functions..applicationId`. -==== Overriding the default binding names generated by the binder with the functional style +== Overriding the default binding names generated by the binder with the functional style By default, the binder uses the strategy discussed above to generate the binding name when using the functional style, i.e. -|-[0..n], for e.g. process-in-0, process-out-0 etc. If you want to override those binding names, you can do that by specifying the following properties. @@ -555,7 +555,7 @@ After that, you must set all the binding level properties on these new binding n Please keep in mind that with the functional programming model described above, adhering to the default binding names make sense in most situations. The only reason you may still want to do this overriding is when you have larger number of configuration properties and you want to map the bindings to something more domain friendly. -==== Setting up bootstrap server configuration +== Setting up bootstrap server configuration When running Kafka Streams applications, you must provide the Kafka broker server information. If you don't provide this information, the binder expects that you are running the broker at the default `localhost:9092`. @@ -567,13 +567,13 @@ If that is not the case, then you need to override that. There are a couple of w When it comes to the binder level property, it doesn't matter if you use the broker property provided through the regular Kafka binder - `spring.cloud.stream.kafka.binder.brokers`. Kafka Streams binder will first check if Kafka Streams binder specific broker property is set (`spring.cloud.stream.kafka.streams.binder.brokers`) and if not found, it looks for `spring.cloud.stream.kafka.binder.brokers`. -=== Record serialization and deserialization +== Record serialization and deserialization Kafka Streams binder allows you to serialize and deserialize records in two ways. One is the native serialization and deserialization facilities provided by Kafka and the other one is the message conversion capabilities of Spring Cloud Stream framework. Lets look at some details. -==== Inbound deserialization +=== Inbound deserialization Keys are always deserialized using native Serdes. @@ -650,7 +650,7 @@ You can use custom message converters by using the following property and an app spring.cloud.stream.bindings.process-in-0.contentType ``` -==== Outbound serialization +=== Outbound serialization Outbound serialization pretty much follows the same rules as above for inbound deserialization. As with the inbound deserialization, one major change from the previous versions of Spring Cloud Stream is that the serialization on the outbound is handled by Kafka natively. @@ -704,14 +704,14 @@ It is worth to mention that the data de/serialization approaches outlined above Your business logic might still need to call Kafka Streams API's that explicitly need `Serde` objects. Those are still the responsibility of the application and must be handled accordingly by the developer. -=== Error Handling +== Error Handling Apache Kafka Streams provides the capability for natively handling exceptions from deserialization errors. For details on this support, please see https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+deserialization+exception+handlers[this]. Out of the box, Apache Kafka Streams provides two kinds of deserialization exception handlers - `LogAndContinueExceptionHandler` and `LogAndFailExceptionHandler`. 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. `LogAndFailExceptionHandler` is the default deserialization exception handler. -==== Handling Deserialization Exceptions in the Binder +=== Handling Deserialization Exceptions in the Binder Kafka Streams binder allows to specify the deserialization exception handlers above using the following property. @@ -777,7 +777,7 @@ If the application is not using either of the above strategies, then it will cre For instance, if your binding's destination topic is `inputTopic` and the application ID is `process-applicationId`, then the default DLQ topic is `error.inputTopic.process-applicationId`. It is always recommended to explicitly create a DLQ topic for each input binding if it is your intention to enable DLQ. -==== DLQ per input consumer binding +=== DLQ per input consumer binding The property `spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler` is applicable for the entire application. This implies that if there are multiple functions in the same application, this property is applied to all of them. @@ -799,7 +799,7 @@ and you only want to enable DLQ on the first input binding and skipAndContinue o Setting deserialization exception handlers this way has a higher precedence than setting at the binder level. -==== DLQ partitioning +=== DLQ partitioning By default, records are published to the Dead-Letter topic using the same partition as the original record. This means the Dead-Letter topic must have at least as many partitions as the original record. @@ -828,12 +828,12 @@ A couple of things to keep in mind when using the exception handling feature in This implies that if there are multiple functions in the same application, this property is applied to all of them. * The exception handling for deserialization works consistently with native deserialization and framework provided message conversion. -==== Handling Production Exceptions in the Binder +=== Handling Production Exceptions in the Binder Unlike the support for deserialization exception handlers as described above, the binder does not provide such first class mechanisms for handling production exceptions. However, you still can configure production exception handlers using the `StreamsBuilderFactoryBean` customizer which you can find more details about, in a subsequent section below. -==== Runtime Error Handling +=== Runtime Error Handling When it comes to handling errors from application code, i.e. from the business logic execution, it is usually up to the application to handle that. Because, the Kafka Streams binder does not have a way to interfere with the application code. @@ -893,7 +893,7 @@ public java.util.function.Function, KStream` to dictate the timestamp used in the record passed in to the `BiConsumer` recoverer. -=== Retrying critical business logic +== Retrying critical business logic There are scenarios in which you might want to retry parts of your business logic that are critical to the application. There maybe an external call to a relational database or invoking a REST endpoint from the Kafka Streams processor. @@ -997,7 +997,7 @@ retryTemplate.execute(context -> { ``` Refer to the https://github.com/spring-projects/spring-retry[Spring Retry] project for more information about the RetryTemplate, retry policies, backoff policies and more. -=== State Store +== State Store State stores are created automatically by Kafka Streams when the high level DSL is used and appropriate calls are made those trigger a state store. @@ -1065,7 +1065,7 @@ Processor() { This will not work when it comes to registering global state stores. In order to register a global state store, please see the section below on customizing `StreamsBuilderFactoryBean`. -=== Interactive Queries +== Interactive Queries Kafka Streams binder API exposes a class called `InteractiveQueryService` to interactively query the state stores. 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. @@ -1124,7 +1124,7 @@ For more information on these host finding methods, please see the Javadoc on th For these methods also, during startup, if the underlying KafkaStreams objects are not ready, they might throw exceptions. The aforementioned retry properties are applicable for these methods as well. -==== Other API methods available through the InteractiveQueryService +=== Other API methods available through the InteractiveQueryService Use the following API method to retrieve the `KeyQueryMetadata` object associated with the combination of given store and key. @@ -1138,7 +1138,7 @@ Use the following API method to retrieve the `KakfaStreams` object associated wi public KafkaStreams getKafkaStreams(String store, K key, Serializer serializer) ``` -==== Customizing Store Query Parameters +=== Customizing Store Query Parameters Sometimes it is necessary that you need to fine tune the store query parameters before querying the store through `InteractiveQueryService`. For this purpose, starting with the `4.0.1` version of the binder, you can provide a bean for `StoreQueryParametersCustomizer` which is a functional interface with a `customize` method that takes a `StoreQueryParameter` as the argument. @@ -1154,7 +1154,7 @@ When this bean is present in this application, `InteractiveQueryService` will ca NOTE: Keep in mind that, there must be a unique bean for `StoreQueryParametersCustomizer` available in the application. -=== Health Indicator +== Health Indicator The health indicator requires the dependency `spring-boot-starter-actuator`. For maven use: [source,xml] @@ -1187,7 +1187,7 @@ When enabling `show-details`, some of the information reported may be redundant. When there are multiple Kafka Streams processors present in the same application, then the health checks will be reported for all of them and will be categorized by the application ID of Kafka Streams. -=== Accessing Kafka Streams Metrics +== Accessing Kafka Streams Metrics Spring Cloud Stream Kafka Streams binder provides Kafka Streams metrics which can be exported through a Micrometer `MeterRegistry`. @@ -1197,7 +1197,7 @@ For Spring Boot version 2.3.x, the Kafka Streams metrics support is provided nat When accessing metrics through the Boot actuator endpoint, make sure to add `metrics` to the property `management.endpoints.web.exposure.include`. Then you can access `/acutator/metrics` to get a list of all the available metrics, which then can be individually accessed through the same URI (`/actuator/metrics/`). -=== Mixing high level DSL and low level Processor API +== Mixing high level DSL and low level Processor API Kafka Streams provides two variants of APIs. It has a higher level DSL like API where you can chain various operations that maybe familiar to a lot of functional programmers. @@ -1259,7 +1259,7 @@ public Consumer> process() { The `process` API method call is a terminal operation while the `transform` API is non terminal and gives you a potentially transformed `KStream` using which you can continue further processing using either the DSL or the processor API. -=== Partition support on the outbound +== Partition support on the outbound A Kafka Streams processor usually sends the processed output into an outbound Kafka topic. If the outbound topic is partitioned and the processor needs to send the outgoing data into particular partitions, the applications needs to provide a bean of type `StreamPartitioner`. @@ -1317,7 +1317,7 @@ spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitio Each output topic in the application needs to be configured separately like this. -=== StreamsBuilderFactoryBean Additional Customizations +== StreamsBuilderFactoryBean Additional Customizations It is often required to customize the `StreamsBuilderFactoryBean` that creates the `KafkaStreams` objects. Based on the underlying support provided by Spring Kafka, the binder allows you to customize the `StreamsBuilderFactoryBean`. @@ -1384,7 +1384,7 @@ public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() }; ``` -==== Using Customizer to register a global state store +== Using Customizer to register a global state store As mentioned above, the binder does not provide a first class way to register global state stores as a feature. For that, you need to use the customizer. @@ -1407,7 +1407,7 @@ public StreamsBuilderFactoryBeanCustomizer customizer() { Again, if you have multiple processors, you want to attach the global state store to the right `StreamsBuilder` by filtering out the other `StreamsBuilderFactoryBean` objects using the application id as outlined above. -==== Using customizer to register a production exception handler +== Using customizer to register a production exception handler In the error handling section, we indicated that the binder does not provide a first class way to deal with production exceptions. Though that is the case, you can still use the `StreamsBuilderFacotryBean` customizer to register production exception handlers. See below. @@ -1425,7 +1425,7 @@ public StreamsBuilderFactoryBeanCustomizer customizer() { Once again, if you have multiple processors, you may want to set it appropriately against the correct `StreamsBuilderFactoryBean`. You may also add such production exception handlers using the configuration property (See below for more on that), but this is an option if you choose to go with a programmatic approach. -=== Timestamp extractor +== Timestamp extractor Kafka Streams allows you to control the processing of the consumer records based on various notions of timestamp. By default, Kafka Streams extracts the timestamp metadata embedded in the consumer record. @@ -1458,7 +1458,7 @@ spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.timestampExtrac If you skip an input consumer binding for setting a custom timestamp extractor, that consumer will use the default settings. -=== Multi binders with Kafka Streams based binders and regular Kafka Binder +== Multi binders with Kafka Streams based binders and regular Message Channel based Kafka Binder You can have an application where you have both a function/consumer/supplier that is based on the regular Kafka binder and a Kafka Streams based processor. However, you cannot mix both of them within a single function or consumer. @@ -1558,14 +1558,14 @@ spring.cloud.stream.bindings.enrichOrder-out-0.binder=kafka1 #kstream ``` -=== State Cleanup +== State Cleanup By default, no local state is cleaned up when the binding is stopped. This is the same behavior effective from Spring Kafka version 2.7. See https://docs.spring.io/spring-kafka/reference/html/#streams-config[Spring Kafka documentation] for more details. 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. -=== Kafka Streams topology visualization +== Kafka Streams topology visualization Kafka Streams binder provides the following actuator endpoints for retrieving the topology description using which you can visualize the topology using external tools. @@ -1577,7 +1577,7 @@ You need to include the actuator and web dependencies from Spring Boot to access Further, you also need to add `kafkastreamstopology` to `management.endpoints.web.exposure.include` property. By default, the `kafkastreamstopology` endpoint is disabled. -=== Event type based routing in Kafka Streams applications +== Event type based routing in Kafka Streams applications Routing functions available in regular message channel based binders are not supported in Kafka Streams binder. However, Kafka Streams binder still provides routing capabilities through the event type record header on the inbound records. @@ -1619,7 +1619,7 @@ In order to address this issue, you can set the following property on the bindin This way, the application can detect deserialization issues right away when using the event routing feature and can take appropriate handling decisions. -=== Binding visualization and control in Kafka Streams binder +== Binding visualization and control in Kafka Streams binder Starting with version 3.1.2, Kafka Streams binder supports binding visualization and control. The only two lifecycle phases supported are `STOPPED` and `STARTED`. @@ -1731,7 +1731,7 @@ When there are multiple bindings present on a single function, invoking these op This is because all the bindings on a single function are backed by the same `StreamsBuilderFactoryBean`. Therefore, for the function above, either `function-in-0` or `function-out-0` will work. -=== Manually starting Kafka Streams processors +== Manually starting Kafka Streams processors Spring Cloud Stream Kafka Streams binder offers an abstraction called `StreamsBuilderFactoryManager` on top of the `StreamsBuilderFactoryBean` from Spring for Apache Kafka. This manager API is used for controlling the multiple `StreamsBuilderFactoryBean` per processor in a binder based application. @@ -1755,7 +1755,7 @@ In order to correct this, you can set `spring.kafka.streams.auto-startup` to `fa Keep in mind that, when using the Spring Cloud Stream binder, you are not directly dealing with `StreamsBuilderFactoryBean` from Spring for Apache Kafka, rather `StreamsBuilderFactoryManager`, as the `StreamsBuilderFactoryBean` objects are internally managed by the binder. -=== Manually starting Kafka Streams processors selectively +== Manually starting Kafka Streams processors selectively While the approach laid out above will unconditionally apply auto start `false` to all the Kafka Streams processors in the application through `StreamsBuilderFactoryManager`, it is often desirable that only individually selected Kafka Streams processors are not auto started. For instance, let us assume that you have three different functions (processors) in your application and for one of the processors, you do not want to start it as part of the application startup. @@ -1827,7 +1827,7 @@ See https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/sprin NOTE: When controlling the bindings by disabling `auto-startup` as described in this section, please note that this is only available for consumer bindings. In other words, if you use the producer binding, `process3-out-0`, that does not have any effect in terms of disabling the auto starting of the processor, although this producer binding uses the same `StreamsBuilderFactoryBean` as the consumer bindings. -=== Tracing using Spring Cloud Sleuth +== Tracing using Spring Cloud Sleuth When Spring Cloud Sleuth is on the classpath of a Spring Cloud Stream Kafka Streams binder based application, both its consumer and producer are automatically instrumented with tracing information. However, in order to trace any application specific operations, those need to be explicitly instrumented by the user code. @@ -1890,13 +1890,13 @@ public Function, KStream> process(KafkaS } ``` -=== Configuration Options +== Configuration Options This section contains the configuration options used by the Kafka Streams binder. For common configuration options and properties pertaining to binder, refer to the <>. -==== Kafka Streams Binder Properties +=== Kafka Streams Binder Properties The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` Any Kafka binder provided properties re-used in Kafka Streams binder must be prefixed with `spring.cloud.stream.kafka.streams.binder` instead of `spring.cloud.stream.kafka.binder`. @@ -2083,7 +2083,7 @@ Custom name for the source component from which the processor is consuming from. + Deafult: `none` (generated by Kafka Streams) -==== Special note on concurrency +== Special note on concurrency In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property. This, you can do using the various `configuration` options described above under binder, functions, producer or consumer level. @@ -2093,4 +2093,4 @@ When you have more than one input binding, set this on the first input binding. For e.g. when setting `spring.cloud.stream.bindings.process-in-0.consumer.concurrency`, it will be translated as `num.stream.threads` by the binder. If you have multiple processors and one processor defines binding level concurrency, but not the others, those ones with no binding level concurrency will default back to the binder wide property specified through `spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads`. -If this binder configuration is not available, then the application will use the default set by Kafka Streams. \ No newline at end of file +If this binder configuration is not available, then the application will use the default set by Kafka Streams. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index 85b386157..53e0a8cd0 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc @@ -1,9 +1,6 @@ -[partintro] --- This guide describes the Apache Kafka implementation of the Spring Cloud Stream Binder. It contains information about its design, usage, and configuration options, as well as information on how the Stream Cloud Stream concepts map onto Apache Kafka specific constructs. In addition, this guide explains the Kafka Streams binding capabilities of Spring Cloud Stream. --- == Apache Kafka Binder @@ -33,8 +30,8 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown The following image shows a simplified diagram of how the Apache Kafka binder operates: -.Kafka Binder -image::{github-raw}/docs/src/main/asciidoc/images/kafka-binder.png[width=300,scaledwidth="50%"] +// .Kafka Binder +// image::{github-raw}/docs/src/main/asciidoc/images/kafka-binder.png[width=300,scaledwidth="50%"] The Apache Kafka Binder implementation maps each destination to an Apache Kafka topic. The consumer group maps directly to the same Apache Kafka concept. @@ -1025,7 +1022,7 @@ Here is the pseudo-code for writing a custom Kafka binder HealthIndicator. In this example, we try to override the binder provided Kafka HealthIndicator by specifically checking first for cluster connectivity and then followed by topic-related issues. -1. First we need create a custom implementation of the `KafkaBinderHealth` interface. +First, we need create a custom implementation of the `KafkaBinderHealth` interface. ``` public class KafkaBinderHealthImplementation implements KafkaBinderHealth { @@ -1062,7 +1059,7 @@ public class KafkaBinderHealthImplementation implements KafkaBinderHealth { } ``` -2. Then we need to create a bean for the custom implementation. +Then we need to create a bean for the custom implementation. ``` @Configuration @@ -1072,4 +1069,4 @@ public class KafkaBinderHealthIndicatorConfiguration { return new KafkaBinderHealthImplementation(admin); } } -``` \ No newline at end of file +``` diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc index e3067edfa..3aa1f4bb3 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc @@ -5,7 +5,7 @@ Here is the pseudo-code for writing a custom Kafka binder HealthIndicator. In this example, we try to override the binder provided Kafka HealthIndicator by specifically checking first for cluster connectivity and then followed by topic-related issues. -1. First we need create a custom implementation of the `KafkaBinderHealth` interface. +First, we need create a custom implementation of the `KafkaBinderHealth` interface. ``` public class KafkaBinderHealthImplementation implements KafkaBinderHealth { @@ -42,7 +42,8 @@ public class KafkaBinderHealthImplementation implements KafkaBinderHealth { } ``` -2. Then we need to create a bean for the custom implementation. + +Then we need to create a bean for the custom implementation. ``` @Configuration diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc index 387bf5ad1..a985f89e9 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc @@ -4,8 +4,8 @@ The following image shows a simplified diagram of how the Apache Kafka binder operates: -.Kafka Binder -image::{github-raw}/docs/src/main/asciidoc/images/kafka-binder.png[width=300,scaledwidth="50%"] +// .Kafka Binder +// image::{github-raw}/docs/src/main/asciidoc/images/kafka-binder.png[width=300,scaledwidth="50%"] The Apache Kafka Binder implementation maps each destination to an Apache Kafka topic. The consumer group maps directly to the same Apache Kafka concept. diff --git a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc index cbcaeed63..1b16b8666 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc @@ -87,12 +87,12 @@ spring.cloud.stream: [[dlq-with-advanced-retry-options]] == DLQ with Advanced Retry Options -[[problem-statement]] +[[dlq-with-advanced-retry-options-problem-statement]] === Problem Statement This is similar to the recipe above, but as a developer I would like to configure the way retries are handled. -[[solution]] +[[dlq-with-advanced-retry-options-solution]] === Solution If you followed the above recipe, then you get the default retry options built into the Kafka binder when the processing encounters an error. @@ -134,14 +134,14 @@ spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name=` by the binders where as you can see the payload of the message maintains its raw form. The headers of the message are ``, where values are typically another primitive or a collection/array of primitives, hence Object. +From the broker, the message arrives in a form of a byte[]. It is then transformed to a `Message` by the binders whereas you can see the payload of the message maintains its raw form. The headers of the message are ``, where values are typically another primitive or a collection/array of primitives, hence Object. That is because binder does not know the required input type as it has no access to the user code (function). So effectively binder delivered an envelope with the payload and some readable meta-data in the form of message headers, just like the letter delivered by mail. This means that while accessing payload of the message is possible you will only have access to it as raw data (i.e., byte[]). And while it may be very common for developers to ask for ability to have SpEL access to fields of a payload object as concrete type (e.g., Foo, Bar etc), you can see how difficult or even impossible would it be to achieve. Here is one example to demonstrate the problem; Imagine you have a routing expression to route to different functions based on payload type. This requirement would imply payload conversion from byte[] to a specific type and then applying the SpEL. However, in order to perform such conversion we would need to know the actual type to pass to converter and that comes from function's signature which we don’t know which one. A better approach to solve this requirement would be to pass the type information as message headers (e.g., `application/json;type=foo.bar.Baz` ). You’ll get a clear readable String value that could be accessed and evaluated in a year and easy to read SpEL expression. -Additionally it is considered very bad practice to use payload for routing decisions, since the payload is considered to be privileged data - data only to be read by its final recipient. Again, using the mail delivery analogy you would not want the mailman to open your envelope and read the contents of the letter to make some delivery decisions. The same concept applies here, especially when it is relatively easy to include such information when generating a Message. It enforces certain level of discipline related to the design of data to be transmitted over the network and which pieces of such data can be considered as public and which are privileged. +Additionally, it is considered very bad practice to use payload for routing decisions, since the payload is considered to be privileged data - data only to be read by its final recipient. Again, using the mail delivery analogy you would not want the mailman to open your envelope and read the contents of the letter to make some delivery decisions. The same concept applies here, especially when it is relatively easy to include such information when generating a Message. It enforces certain level of discipline related to the design of data to be transmitted over the network and which pieces of such data can be considered as public and which are privileged. diff --git a/docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc b/docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc index cee7b3b12..92c0f92e9 100644 --- a/docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc +++ b/docs/modules/ROOT/pages/pulsar/pulsar_binder.adoc @@ -19,7 +19,6 @@ Maven:: org.springframework.pulsar spring-pulsar-spring-cloud-stream-binder - {spring-pulsar-version} ---- @@ -29,7 +28,7 @@ Gradle:: [source,groovy,indent=0,subs="verbatim,attributes",role="secondary"] ---- dependencies { - implementation 'org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:{spring-pulsar-version}' + implementation 'org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder' } ---- ====== @@ -39,7 +38,7 @@ dependencies { The Spring Cloud Stream binder for Apache Pulsar allows the applications to focus on business logic rather than dealing with the lower-level details of managing and maintaining Pulsar. The binder takes care of all those details for the application developer. -Spring Cloud Stream brings a powerful programming model based on {spring-cloud-function}[Spring Cloud Function] that allows the app developer to write complex event-driven applications using a functional style. +Spring Cloud Stream brings a powerful programming model based on https://spring.io/projects/spring-cloud-function[Spring Cloud Function] that allows the app developer to write complex event-driven applications using a functional style. Applications can start from a middleware-neutral manner and then map Pulsar topics as destinations in Spring Cloud Stream through Spring Boot configuration properties. Spring Cloud Stream is built on top of Spring Boot, and when writing an event-driven microservice using Spring Cloud Stream, you are essentially writing a Boot application. Here is a straightforward Spring Cloud Stream application. diff --git a/docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc b/docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc index d92490de0..36a215446 100644 --- a/docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc +++ b/docs/modules/ROOT/pages/pulsar/spring-cloud-stream-binder-pulsar.adoc @@ -14,9 +14,6 @@ Soby Chacko; Chris Bono; Alexander Preuß; Jay Bryant; Christophe Bornet // ====================================================================================== -*{project-version}* - -[[reference-guide]] -= Reference Guide -:page-section-summary-toc: 1 +// *{project-version}* +xref:pulsar/pulsar_binder.adoc[Reference Guide] diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc index b13baf694..6290ac379 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_overview.adoc @@ -1,11 +1,9 @@ -[partintro] --- This guide describes the RabbitMQ implementation of the Spring Cloud Stream Binder. It contains information about its design, usage and configuration options, as well as information on how the Stream Cloud Stream concepts map into RabbitMQ specific constructs. --- + [[usage]] -= Usage +== Usage To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates: @@ -28,12 +26,12 @@ Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows: ---- [[rabbitmq-binder-overview]] -= RabbitMQ Binder Overview +== RabbitMQ Binder Overview The following simplified diagram shows how the RabbitMQ binder operates: -.RabbitMQ Binder -image::{github-raw}/docs/src/main/asciidoc/images/rabbit-binder.png[width=300,scaledwidth="50%"] +// .RabbitMQ Binder +// image::{github-raw}/docs/src/main/asciidoc/images/rabbit-binder.png[width=300,scaledwidth="50%"] By default, the RabbitMQ Binder implementation maps each destination to a `TopicExchange`. For each consumer group, a `Queue` is bound to that `TopicExchange`. @@ -77,7 +75,7 @@ Starting with version 2.0, the `RabbitMessageChannelBinder` sets the `RabbitTemp NOTE: Currently, a `multiplex` consumer (a single consumer listening to multiple queues) is only supported for message-driven consumers; polled consumers can only retrieve messages from a single queue. [[configuration-options]] -= Configuration Options +== Configuration Options This section contains settings specific to the RabbitMQ Binder and bound channels. diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_overview/putting-it-all-together.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_overview/putting-it-all-together.adoc index 291e55756..e154a8bbf 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_overview/putting-it-all-together.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_overview/putting-it-all-together.adoc @@ -47,7 +47,7 @@ public class XDeathApplication { Notice that the count property in the `x-death` header is a `Long`. [[rabbit-error-channels]] -= Error Channels +== Error Channels 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 "`xref:spring-cloud-stream/overview-error-handling.adoc[Error Handling]`" for more information. @@ -91,7 +91,7 @@ There is no automatic handling of these exceptions (such as sending to a xref:ra You can consume these exceptions with your own Spring Integration flow. [[rabbit-binder-health-indicator]] -= Rabbit Binder Health Indicator +== Rabbit Binder Health Indicator The health indicator for Rabbit binder delegates to the one provided from Spring Boot. For more information on this, see https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#actuator.endpoints.health.auto-configured-health-indicators[this]. diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_overview/rabbitmq-stream-producer.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_overview/rabbitmq-stream-producer.adoc index 5d6cc684f..f8d41bd4e 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_overview/rabbitmq-stream-producer.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_overview/rabbitmq-stream-producer.adoc @@ -91,7 +91,7 @@ Refer to the property documentation above for more information. There are similar properties used when declaring a dead-letter exchange/queue, when `autoBindDlq` is `true`. [[retry-with-the-rabbitmq-binder]] -= Retry With the RabbitMQ Binder +== Retry With the RabbitMQ Binder 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. diff --git a/docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc b/docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc index b6f2d9308..fd6e4b98f 100644 --- a/docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc +++ b/docs/modules/ROOT/pages/rabbit/spring-cloud-stream-binder-rabbit.adoc @@ -27,12 +27,6 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat // ====================================================================================== -*{project-version}* - - -[[reference-guide]] -= Reference Guide - - - +// *{project-version}* +xref:rabbit/rabbit_overview.adoc[Reference Guide] diff --git a/docs/src/main/asciidoc/sagan-index.adoc b/docs/modules/ROOT/pages/sagan-index.adoc similarity index 100% rename from docs/src/main/asciidoc/sagan-index.adoc rename to docs/modules/ROOT/pages/sagan-index.adoc diff --git a/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc b/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc index 9aac7bf0a..0f3b2e41a 100644 --- a/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc +++ b/docs/modules/ROOT/pages/schema-registry/spring-cloud-stream-schema-registry.adoc @@ -11,7 +11,7 @@ Spring Cloud Stream schema registry support also provides support for avro based The schema evolution support provided by Spring Cloud Stream works both with the aforementioned standalone schema registry as well as the schema registry provided by Confluent that specifically works with Apache Kafka. [[spring-cloud-stream-schema-registry-overview]] -==== Spring Cloud Stream Schema Registry overview +== Spring Cloud Stream Schema Registry overview Spring Cloud Stream Schema Registry 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. @@ -38,7 +38,7 @@ Spring Cloud Stream Schema Registry provides the following components Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry. [[schema-registry-client]] -=== Schema Registry Client +== Schema Registry Client The client-side abstraction for interacting with schema registry servers is the `SchemaRegistryClient` interface, which has the following structure: @@ -74,7 +74,7 @@ If you intend to change the default behavior, you can use the client directly on To do so, you have to add the property `spring.cloud.stream.schemaRegistryClient.cached=true` to your application properties. [[schema-registry-client-properties]] -==== Schema Registry Client Properties +== Schema Registry Client Properties The Schema Registry Client supports the following properties: @@ -234,20 +234,20 @@ You can customize the schema storage by using the http://docs.spring.io/spring-b [[schema-registry-server-api]] -==== Schema Registry Server API +== Schema Registry Server API The Schema Registry Server API consists of the following operations: * `POST /` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-registering-new-schema[Registering a New Schema]` -* `GET /{subject}/{format}/{version}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format-version[Retrieving an Existing Schema by Subject, Format, and Version]` -* `GET /{subject}/{format}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format[Retrieving an Existing Schema by Subject and Format]` -* `GET /schemas/{id}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-id[Retrieving an Existing Schema by ID]` -* `DELETE /{subject}/{format}/{version}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject-format-version[Deleting a Schema by Subject, Format, and Version]` -* `DELETE /schemas/{id}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-id[Deleting a Schema by ID]` -* `DELETE /{subject}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject[Deleting a Schema by Subject]` +* `GET /\{subject\}/\{format\}/\{version\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format-version[Retrieving an Existing Schema by Subject, Format, and Version]` +* `GET /\{subject\}/\{format\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format[Retrieving an Existing Schema by Subject and Format]` +* `GET /schemas/\{id\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-id[Retrieving an Existing Schema by ID]` +* `DELETE /\{subject\}/\{format\}/\{version\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject-format-version[Deleting a Schema by Subject, Format, and Version]` +* `DELETE /schemas/\{id\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-id[Deleting a Schema by ID]` +* `DELETE /\{subject\}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject[Deleting a Schema by Subject]` [[spring-cloud-stream-overview-registering-new-schema]] -===== Registering a New Schema +== Registering a New Schema To register a new schema, send a `POST` request to the `/` endpoint. @@ -266,9 +266,9 @@ Its response is a schema object in JSON, with the following fields: * `definition`: The schema definition [[spring-cloud-stream-overview-retrieve-schema-subject-format-version]] -===== Retrieving an Existing Schema by Subject, Format, and Version +== Retrieving an Existing Schema by Subject, Format, and Version -To retrieve an existing schema by subject, format, and version, send `GET` request to the `/{subject}/{format}/{version}` endpoint. +To 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: @@ -279,7 +279,7 @@ Its response is a schema object in JSON, with the following fields: * `definition`: The schema definition [[spring-cloud-stream-overview-retrieve-schema-subject-format]] -===== Retrieving an Existing Schema by Subject and Format +== Retrieving an Existing Schema by Subject and Format To retrieve an existing schema by subject and format, send a `GET` request to the `/subject/format` endpoint. @@ -292,9 +292,9 @@ Its response is a list of schemas with each schema object in JSON, with the foll * `definition`: The schema definition [[spring-cloud-stream-overview-retrieve-schema-id]] -===== Retrieving an Existing Schema by ID +== Retrieving an Existing Schema by ID -To retrieve a schema by its ID, send a `GET` request to the `/schemas/{id}` endpoint. +To 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: @@ -305,18 +305,18 @@ Its response is a schema object in JSON, with the following fields: * `definition`: The schema definition [[spring-cloud-stream-overview-deleting-schema-subject-format-version]] -===== Deleting a Schema by Subject, Format, and Version +== Deleting a Schema by Subject, Format, and Version -To delete a schema identified by its subject, format, and version, send a `DELETE` request to the `/{subject}/{format}/{version}` endpoint. +To delete a schema identified by its subject, format, and version, send a `DELETE` request to the `\{subject\}/\{format\}/\{version\}` endpoint. [[spring-cloud-stream-overview-deleting-schema-id]] -===== Deleting a Schema by ID +== Deleting a Schema by ID -To delete a schema by its ID, send a `DELETE` request to the `/schemas/{id}` endpoint. +To delete a schema by its ID, send a `DELETE` request to the `/schemas/\{id\}` endpoint. [[spring-cloud-stream-overview-deleting-schema-subject]] -===== Deleting a Schema by Subject -`DELETE /{subject}` +== Deleting a Schema by Subject +`DELETE /\{subject\}` Delete existing schemas by their subject. @@ -326,7 +326,7 @@ To avoid any conflicts in the future, starting with 1.1.1.RELEASE, we have opted Any Spring Cloud Stream 1.1.0.RELEASE users who upgrade should migrate their existing schemas to the new table before upgrading. [[using-confluent-s-schema-registry]] -==== Using Confluent's Schema Registry +== Using Confluent's Schema Registry 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: @@ -343,7 +343,7 @@ public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.s NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform version 4.0.0. [[schema-registration-and-resolution]] -=== Schema Registration and Resolution +== Schema Registration and Resolution To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections: @@ -351,32 +351,32 @@ To better understand how Spring Cloud Stream registers and resolves new schemas * `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-schema-resolution-process[Schema Resolution Process (Deserialization)]` [[spring-cloud-stream-overview-schema-registration-process]] -==== Schema Registration Process (Serialization) +== Schema Registration Process (Serialization) The first part of the registration process is extracting a schema from the payload that is being sent over a channel. Avro types such as `SpecificRecord` or `GenericRecord` already contain a schema, which can be retrieved immediately from the instance. In the case of POJOs, a schema is inferred if the `spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled` property is set to `true` (the default). -.Schema Writer Resolution Process -image::{github-raw}/docs/src/main/asciidoc/images/schema_resolution.png[width=800,scaledwidth="75%",align="center"] +// .Schema Writer Resolution Process +// image::{github-raw}/docs/src/main/asciidoc/images/schema_resolution.png[width=800,scaledwidth="75%",align="center"] 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. -.Schema Registration Process -image::{github-raw}/docs/src/main/asciidoc/images/registration.png[width=800,scaledwidth="75%",align="center"] +// .Schema Registration Process +// image::{github-raw}/docs/src/main/asciidoc/images/registration.png[width=800,scaledwidth="75%",align="center"] 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`. [[spring-cloud-stream-overview-schema-resolution-process]] -==== Schema Resolution Process (Deserialization) +== Schema Resolution Process (Deserialization) When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-schema-registration-process[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). -.Schema Reading Resolution Process -image::{github-raw}/docs/src/main/asciidoc/images/schema_reading.png[width=800,scaledwidth="75%",align="center"] +// .Schema Reading Resolution Process +// image::{github-raw}/docs/src/main/asciidoc/images/schema_reading.png[width=800,scaledwidth="75%",align="center"] 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 https://avro.apache.org/docs/1.7.6/spec.html[the Avro terminology] and understand the process. diff --git a/docs/modules/ROOT/pages/spring-cloud-stream.adoc b/docs/modules/ROOT/pages/spring-cloud-stream.adoc index 87191b369..d0b7e8250 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream.adoc @@ -1,9 +1,6 @@ :nofooter: :sectlinks: true -*{project-version}* - - [[spring-cloud-stream-reference]] :doctype: book @@ -11,17 +8,14 @@ // ====================================================================================== [[preface]] -= Preface +== Preface -[partintro] --- 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-overview-introducing]] -= Introducing Spring Cloud Stream +== Introducing Spring Cloud Stream Spring Cloud Stream is a framework for building message-driven microservice applications. Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. @@ -72,7 +66,7 @@ class BootTestStreamApplicationTests { ---- [[main-concepts]] -= Main Concepts +== Main Concepts Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications. This section gives an overview of the following: diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/binding-properties.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/binding-properties.adoc index 4b4ebf886..0d72a8291 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/binding-properties.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/binding-properties.adoc @@ -259,7 +259,7 @@ All you need to do is provide implementation of this strategy and configure it a [[content-type-management]] -= Content Type Negotiation +== Content Type Negotiation Data transformation is one of the core features of any message-driven microservice architecture. Given that, in Spring Cloud Stream, such data is represented as a Spring `Message`, a message may have to be transformed to a desired shape or size before reaching its destination. This is required for two reasons: diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/deploying-applications-on-cloudfoundry.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/deploying-applications-on-cloudfoundry.adoc index 0eacad7de..ebd37692f 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/deploying-applications-on-cloudfoundry.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/deploying-applications-on-cloudfoundry.adoc @@ -7,7 +7,7 @@ On CloudFoundry, services are usually exposed through a special environment vari When configuring your binder connections, you can use the values from an environment variable as explained on the http://docs.spring.io/spring-cloud-dataflow-server-cloudfoundry/docs/current-SNAPSHOT/reference/htmlsingle/#getting-started-ups[dataflow Cloud Foundry Server] docs. [[binder-implementations]] -= Binder Implementations +== Binder Implementations :page-section-summary-toc: 1 The following is the list of available binder implementations diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc index e7494d2b8..43c49f05f 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/overview-custom-binder-impl.adoc @@ -246,7 +246,7 @@ com.example.springcloudstreamcustombinder.config.FileMessageBinderConfiguration ---- [[configuration-options]] -= Configuration Options +== Configuration Options Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders. Some binders let additional binding properties support middleware-specific features. diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc index f89c65fe6..9c9f2dfe8 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc @@ -210,7 +210,7 @@ spring.cloud.stream.bindings..consumer.retry-template-name=io.spring.maven.antora antora-maven-plugin - - org.apache.maven.plugins - maven-antrun-plugin - + + + + maven-deploy-plugin diff --git a/docs/src/main/antora/resources/antora-resources/antora.yml b/docs/src/main/antora/resources/antora-resources/antora.yml index 9148923fa..0b93ee6df 100644 --- a/docs/src/main/antora/resources/antora-resources/antora.yml +++ b/docs/src/main/antora/resources/antora-resources/antora.yml @@ -1,20 +1,20 @@ version: @antora-component.version@ prerelease: @antora-component.prerelease@ -asciidoc: - attributes: - attribute-missing: 'warn' - chomp: 'all' - project-root: @maven.multiModuleProjectDirectory@ - github-repo: @docs.main@ - github-raw: https://raw.githubusercontent.com/spring-cloud/@docs.main@/@github-tag@ - github-code: https://github.com/spring-cloud/@docs.main@/tree/@github-tag@ - github-issues: https://github.com/spring-cloud/@docs.main@/issues/ - github-wiki: https://github.com/spring-cloud/@docs.main@/wiki - spring-cloud-version: @project.version@ - github-tag: @github-tag@ - version-type: @version-type@ - docs-url: https://docs.spring.io/@docs.main@/docs/@project.version@ - raw-docs-url: https://raw.githubusercontent.com/spring-cloud/@docs.main@/@github-tag@ - project-version: @project.version@ - project-name: @docs.main@ +#asciidoc: +# attributes: +# attribute-missing: 'warn' +# chomp: 'all' +# project-root: @maven.multiModuleProjectDirectory@ +# github-repo: @docs.main@ +# github-raw: https://raw.githubusercontent.com/spring-cloud/@docs.main@/@github-tag@ +# github-code: https://github.com/spring-cloud/@docs.main@/tree/@github-tag@ +# github-issues: https://github.com/spring-cloud/@docs.main@/issues/ +# github-wiki: https://github.com/spring-cloud/@docs.main@/wiki +# spring-cloud-version: @project.version@ +# github-tag: @github-tag@ +# version-type: @version-type@ +# docs-url: https://docs.spring.io/@docs.main@/docs/@project.version@ +# raw-docs-url: https://raw.githubusercontent.com/spring-cloud/@docs.main@/@github-tag@ +# project-version: @project.version@ +# project-name: @docs.main@