From 1ee5247fd73c67c44177f62165bd1381a5897a96 Mon Sep 17 00:00:00 2001 From: Tran Ngoc Nhan Date: Sat, 31 Oct 2026 20:33:16 +0700 Subject: [PATCH] Convert to ASCII syntax Signed-off-by: Tran Ngoc Nhan --- README.adoc | 7 +- .../admin-client-config-cust.adoc | 5 +- .../kafka/kafka-binder/consume-batches.adoc | 5 +- .../kafka/kafka-binder/custom-health-ind.adoc | 15 +- .../pages/kafka/kafka-binder/metrics.adoc | 5 +- .../kafka/kafka-binder/security-config.adoc | 5 +- .../kafka-reactive-binder/concurrency.adoc | 5 +- .../kafka-reactive-binder/consuming.adoc | 7 +- .../reactive_observability.adoc | 15 +- .../kafka/kafka-reactive-binder/usage.adoc | 5 +- .../ancillaries-to-the-programming-model.adoc | 15 +- ...g-visualization-and-control-in-binder.adoc | 30 ++- .../kafka-streams-binder/error-handling.adoc | 25 +- ...nt-type-based-routing-in-applications.adoc | 5 +- .../interactive-queries.adoc | 15 +- ...ually-starting-processors-selectively.adoc | 25 +- .../manually-starting-processors.adoc | 5 +- ...level-dsl-and-low-level-processor-api.adoc | 10 +- ...with-based-binders-and-regular-binder.adoc | 63 ++--- .../partition-support-on-the-outbound.adoc | 20 +- .../programming-model.adoc | 56 ++-- .../retrying-critical-business-logic.adoc | 22 +- .../streamsbuilderfactorybean-customizer.adoc | 26 +- .../timestamp-extractor.adoc | 12 +- .../tracing-using-spring-cloud-sleuth.adoc | 10 +- .../ROOT/pages/kafka/kafka_overview.adoc | 5 +- docs/modules/ROOT/pages/kafka/kafka_tips.adoc | 241 +++++++++++------- .../rabbit_overview/health-indicator.adoc | 5 +- tools/kafka/docker-compose/README.adoc | 10 +- 29 files changed, 397 insertions(+), 277 deletions(-) diff --git a/README.adoc b/README.adoc index a5e010cc21..8826767df3 100644 --- a/README.adoc +++ b/README.adoc @@ -122,7 +122,7 @@ build succeed, please raise a ticket to get the settings added to source control. The projects that require middleware (i.e. Redis) for testing generally -require that a local instance of [Docker](https://www.docker.com/get-started) is installed and running. +require that a local instance of https://www.docker.com/get-started[Docker] is installed and running. === Documentation @@ -296,10 +296,11 @@ If you need to suppress some rules (e.g. line length needs to be longer), then i It's advisable to copy the `${spring-cloud-build.rootFolder}/.editorconfig` and `${spring-cloud-build.rootFolder}/.springformat` to your project. That way, some default formatting rules will be applied. You can do so by running this script: -```bash +[source,bash] +---- $ curl https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/.editorconfig -o .editorconfig $ touch .springformat -``` +---- === IDE setup diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc index ef47393920..acb298182a 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc @@ -6,11 +6,12 @@ AdminClientConfigCustomizer's configure method provides access to the admin clie Binder's Kafka topic provisioner gives the highest precedence for the properties given through this customizer. Here is an example of providing this customizer bean. -``` +[source,java] +---- @Bean public AdminClientConfigCustomizer adminClientConfigCustomizer() { return props -> { props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL"); }; } -``` +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc index 86d6866b3f..6b470572c6 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc @@ -6,10 +6,11 @@ The size of the batch is controlled by Kafka consumer properties `max.poll.recor When receiving the batches, the following type signatures are allowed: -``` +[source,java] +---- List Message> -``` +---- In the first option of `List`, the listener will not get any message headers. If the second type signature (`Message>`) is used, then the headers can be accessed; however, all the headers are still be in the form of a `Collection`. diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc index f956901cdc..64d103c50b 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc @@ -12,7 +12,8 @@ The custom implementation must be present in the application configuration as a When the binder discovers the custom implementation, it will use that instead of the default implementation. Here is an example of such a custom implementation bean in the application. -``` +[source,java] +---- @Bean public KafkaBinderHealth kafkaBinderHealthIndicator() { return new KafkaBinderHealth() { @@ -22,7 +23,7 @@ public KafkaBinderHealth kafkaBinderHealthIndicator() { } }; } -``` +---- [[custom-kafka-binder-health-indicator-example]] == Custom kafka Binder Health Indicator Example @@ -33,7 +34,8 @@ In this example, we try to override the binder provided Kafka HealthIndicator by First, we need create a custom implementation of the `KafkaBinderHealth` interface. -``` +[source,java] +---- public class KafkaBinderHealthImplementation implements KafkaBinderHealth { @Value("${spring.cloud.bus.destination}") private String topic; @@ -66,11 +68,12 @@ public class KafkaBinderHealthImplementation implements KafkaBinderHealth { // Your implementation } } -``` +---- Then we need to create a bean for the custom implementation. -``` +[source,java] +---- @Configuration public class KafkaBinderHealthIndicatorConfiguration { @Bean @@ -78,4 +81,4 @@ public class KafkaBinderHealthIndicatorConfiguration { return new KafkaBinderHealthImplementation(admin); } } -``` +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc index cf08ae2447..1855b34c36 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc @@ -14,7 +14,8 @@ refer to the <> for mo You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application. -``` +[source,java] +---- @Component class NoOpBindingMeters { NoOpBindingMeters(MeterRegistry registry) { @@ -22,6 +23,6 @@ class NoOpBindingMeters { MeterFilter.denyNameStartsWith(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)); } } -``` +---- More details on how to suppress meters selectively can be found https://micrometer.io/docs/concepts#_meter_filters[here]. diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc index 6dbe6b5d0b..54dfb2a283 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc @@ -94,7 +94,8 @@ See this https://cwiki.apache.org/confluence/display/KAFKA/KIP-85%3A+Dynamic+JAA For example, if you have two clusters in your application with separate JAAS configuration, then the following is a template that you can use: -``` +[source,yaml] +---- spring.cloud.stream: binders: kafka1: @@ -121,7 +122,7 @@ spring.cloud.stream: configuration: security.protocol: SASL_PLAINTEXT sasl.mechanism: PLAIN -``` +---- Note that both the Kafka clusters, and the `sasl.jaas.config` values for each of them are different in the above configuration. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc index 46f74aef4c..d356fe8402 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc @@ -8,8 +8,9 @@ This could be useful when you are consuming records from a partitioned topic. For example, assume that the incoming topic has at least three partitions. Then you can set the following property. -``` +[source,properties] +---- spring.cloud.stream.bindings.lowercase-in-0.consumer.concurrency=3 -``` +---- That will create three dedicated `KafkaReceiver` objects that generate three separate `Flux` implementations and then stream them to the handler method. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/consuming.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/consuming.adoc index 6bdb6b1d12..e244935e20 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/consuming.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/consuming.adoc @@ -48,9 +48,10 @@ RecordMessageConverter fullRawReceivedRecord() { Then, you need to instruct the framework to use this converter for the required binding. Here is an example based on our `lowercase` function. -``` -spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRawReceivedRecord" -``` +[source,properties] +---- +spring.cloud.stream.kafka.bindings.lowercase-in-0.consumer.converterBeanName=fullRawReceivedRecord +---- `lowercase-in-0` is the input binding name for our `lowercase` function. For the outbound (`lowercase-out-0`), we still use the regular `MessagingMessageConverter`. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc index 50ec81c265..f589cec5f5 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc @@ -8,9 +8,10 @@ This section describes how Micrometer-based observability is enabled in the reac There is built-in support for observability in producer binding. To enable it, set the following property: -``` +[source,properties] +---- spring.cloud.stream.kafka.binder.enable-observation -``` +---- When this property is set to `true`, you can observe the publishing of records. Both publishing records using `StreamBridge` and regular `Supplier` beans can be observed. @@ -28,7 +29,8 @@ In the second case, if there was no upstream observation started, it will start === Example: Function with Observability -``` +[source,java] +---- @Bean Function>, Flux>> receive(ObservationRegistry observationRegistry) { @@ -50,7 +52,7 @@ Function>, Flux>> receive(Ob .contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation)); }); } -``` +---- In this example: @@ -63,7 +65,8 @@ In this example: === Example: Consumer with Observability -``` +[source,java] +---- @Bean Consumer>> receive(ObservationRegistry observationRegistry, @Value("${spring.kafka.bootstrap-servers}") String bootstrap) { return f -> f.doOnNext(record -> KafkaReceiverObservation.RECEIVER_OBSERVATION.observation( @@ -73,7 +76,7 @@ Consumer>> receive(ObservationRegistry observatio observationRegistry).observe(() -> System.out.println(record))) .subscribe(); } -``` +---- In this case: diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc index fed5e0f837..9043f7c240 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc @@ -3,9 +3,10 @@ Following are the maven coordinates for the reactive Kafka binder. -``` +[source,xml] +---- org.springframework.cloud spring-cloud-stream-binder-kafka-reactive -``` +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc index eff340d3ee..a0f2e4ae7e 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc @@ -7,7 +7,8 @@ Binder allows to have multiple Kafka Streams processors within a single Spring Cloud Stream application. You can have an application as below. -``` +[source,java] +---- @Bean public java.util.function.Function, KStream> process() { ... @@ -23,7 +24,7 @@ public java.util.function.BiFunction, KTable> process() { ... } -``` +---- and -``` +[source,java] +---- @Bean public java.util.function.Consumer> anotherProcess() { ... } -``` +---- Then you can set the application id for each, using the following binder level properties. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc index d2823c565c..05889135c0 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc @@ -7,7 +7,8 @@ The lifecycle phases `PAUSED` and `RESUMED` are not available in Kafka Streams b In order to activate binding visualization and control, the application needs to include the following two dependencies. -``` +[source,xml] +---- org.springframework.boot spring-boot-starter-actuator @@ -16,19 +17,21 @@ In order to activate binding visualization and control, the application needs to org.springframework.boot spring-boot-starter-web -``` +---- If you prefer using webflux, you can then include `spring-boot-starter-webflux` instead of the standard web dependency. In addition, you also need to set the following property: -``` +[source,properties] +---- management.endpoints.web.exposure.include=bindings -``` +---- To illustrate this feature further, let us use the following application as a guide: -``` +[source,java] +---- @SpringBootApplication public class KafkaStreamsApplication { @@ -47,7 +50,7 @@ public class KafkaStreamsApplication { } } -``` +---- As we can see, the application has two Kafka Streams functions - one, a consumer and another a function. The consumer binding is named by default as `consumer-in-0`. @@ -55,7 +58,8 @@ Similarly, for the function, the input binding is `function-in-0` and the output Once the application is started, we can find details about the bindings using the following bindings endpoint. -``` +[source,bash] +---- curl http://localhost:8080/actuator/bindings | jq . [ { @@ -89,23 +93,25 @@ Once the application is started, we can find details about the bindings using th "extendedInfo": {} } ] -``` +---- The details about all three bindings can be found above. Let us now stop the consumer-in-0 binding. -``` +[source,bash] +---- curl -d '{"state":"STOPPED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/consumer-in-0 -``` +---- At this point, no records will be received through this binding. Start the binding again. -``` +[source,bash] +---- curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/consumer-in-0 -``` +---- When there are multiple bindings present on a single function, invoking these operations on any of those bindings will work. This is because all the bindings on a single function are backed by the same `StreamsBuilderFactoryBean`. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc index 3fe5bb3893..86f39f4d08 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc @@ -82,12 +82,13 @@ However, if you have multiple processors or multiple input bindings within a sin If you have the following processor, -``` +[source,java] +---- @Bean public BiFunction, KTable, KStream> process() { ... } -``` +---- and you only want to enable DLQ on the first input binding and skipAndContinue on the second binding, then you can do so on the consumer as below. @@ -141,13 +142,14 @@ However, to make things a bit easier for the application, the binder provides a Consider the following code. -``` +[source,java] +---- @Bean public java.util.function.Function, KStream> process() { return input -> input .map(...); } -``` +---- If the business code inside your `map` call above throws an exception, it is your responsibility to handle that error. This is where `RecordRecoverableProcessor` becomes handy. @@ -156,7 +158,8 @@ Let's say that you want to publish the failed record to a DLT, rather than handl In that case, you must use a custom implementation of `RecordRecoverableProcessor` called `DltAwareProcessor`. Here is how you can do that. -``` +[source,java] +---- @Bean public java.util.function.Function, KStream> process(DltPublishingContext dltSenderContext) { return input -> input @@ -164,7 +167,7 @@ public java.util.function.Function, KStream, KStream> process() { return input -> input @@ -189,7 +193,7 @@ public java.util.function.Function, KStream, KStream> process() { return input -> input; } -``` +---- Let us also assume that we only want the business logic in this function to be executed, if the incoming record has event types as `foo` or `bar`. That can be expressed as below using the `eventTypes` property on the binding. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc index 9a97d4914f..8db35926ac 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc @@ -63,15 +63,17 @@ The aforementioned retry properties are applicable for these methods as well. Use the following API method to retrieve the `KeyQueryMetadata` object associated with the combination of given store and key. -``` +[source,java] +---- public KeyQueryMetadata getKeyQueryMetadata(String store, K key, Serializer serializer) -``` +---- Use the following API method to retrieve the `KakfaStreams` object associated with the combination of given store and key. -``` +[source,java] +---- public KafkaStreams getKafkaStreams(String store, K key, Serializer serializer) -``` +---- [[customizing-store-query-parameters]] == Customizing Store Query Parameters @@ -80,9 +82,10 @@ Sometimes it is necessary that you need to fine tune the store query parameters 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. Here is its method signature. -``` +[source,java] +---- StoreQueryParameters customize(StoreQueryParameters storeQueryParameters); -``` +---- Using this approach, applications can further customize the `StoreQueryParameters` such as enabling stale stores. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc index 3b19c7f983..d01f3cc6b4 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc @@ -5,7 +5,8 @@ While the approach laid out above will unconditionally apply auto start `false` 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. Here is an example of such a situation. -``` +[source,java] +---- @Bean public Function, KStream> process1() { @@ -22,7 +23,7 @@ public BiFunction, KTable, KStream> process3() { } -``` +---- In this scenario above, if you set `spring.kafka.streams.auto-startup` to `false`, then none of the processors will auto start during the application startup. In that case, you have to programmatically start them as described above by calling `start()` on the underlying `StreamsBuilderFactoryManager`. @@ -35,26 +36,30 @@ Because they share the same factory bean, you don't have to set autoStartup to f Here is the Spring Cloud Stream property that you can use to disable auto startup for this processor. -``` +[source,yaml] +---- spring.cloud.stream.bindings.process3-in-0.consumer.auto-startup: false -``` +---- or -``` +[source,yaml] +---- spring.cloud.stream.bindings.process3-in-1.consumer.auto-startup: false -``` +---- Then, you can manually start the processor either using the REST endpoint or using the `BindingsEndpoint` API as shown below. For this, you need to ensure that you have the Spring Boot actuator dependency on the classpath. -``` +[source,bash] +---- curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/process3-in-0 -``` +---- or -``` +[source,java] +---- @Autowired BindingsEndpoint endpoint; @@ -64,7 +69,7 @@ public ApplicationRunner runner() { endpoint.changeState("process3-in-0", State.STARTED); }; } -``` +---- See https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#binding_visualization_control[this section] from the reference docs for more details on this mechanism. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc index 7ba74e3b49..d27d9bcedf 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc @@ -8,14 +8,15 @@ Therefore, when using the binder, if you manually want to control the auto start You can use the property `spring.kafka.streams.auto-startup` and set this to `false` in order to turn off auto starting of the processors. Then, in the application, you can use something as below to start the processors using `StreamsBuilderFactoryManager`. -``` +[source,java] +---- @Bean public ApplicationRunner runner(StreamsBuilderFactoryManager sbfm) { return args -> { sbfm.start(); }; } -``` +---- This feature is handy, when you want your application to start in the main thread and let Kafka Streams processors start separately. For example, when you have a large state store that needs to be restored, if the processors are started normally as is the default case, this may block your application to start. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc index 69670c5dd1..a608fbb815 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc @@ -11,7 +11,8 @@ Applications can use the `transform` or `process` method API calls to get access Here is a look at how one may combine both the DSL and the processor API in a Spring Cloud Stream application using the `process` API. -``` +[source,java] +---- @Bean public Consumer> process() { return input -> @@ -32,11 +33,12 @@ public Consumer> process() { }); } -``` +---- Here is an example using the `transform` API. -``` +[source,java] +---- @Bean public Consumer> process() { return (input, a) -> @@ -57,7 +59,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. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc index db2ec8c036..95a07eecd6 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc @@ -6,7 +6,8 @@ However, you cannot mix both of them within a single function or consumer. Here is an example, where you have both binder based components within the same application. -``` +[source,java] +---- @Bean public Function process() { return s -> s; @@ -18,46 +19,48 @@ public Function, KStream> kstreamProcess() return input -> input; } -``` +---- This is the relevant parts from the configuration: -``` +[source,properties] +---- spring.cloud.function.definition=process;kstreamProcess spring.cloud.stream.bindings.process-in-0.destination=foo spring.cloud.stream.bindings.process-out-0.destination=bar spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar -``` +---- Things become a bit more complex if you have the same application as above, but is dealing with two different Kafka clusters, for e.g. the regular `process` is acting upon both Kafka cluster 1 and cluster 2 (receiving data from cluster-1 and sending to cluster-2) and the Kafka Streams processor is acting upon Kafka cluster 2. Then you have to use the https://cloud.spring.io/spring-cloud-stream/reference/html/spring-cloud-stream.html#multiple-binders[multi binder] facilities provided by Spring Cloud Stream. Here is how your configuration may change in that scenario. -``` +[source,yaml] +---- # multi binder configuration spring.cloud.stream.binders.kafka1.type: kafka -spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-1} #Replace kafkaCluster-1 with the approprate IP of the cluster +spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-1} #Replace kafkaCluster-1 with the approprate IP of the cluster spring.cloud.stream.binders.kafka2.type: kafka -spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster +spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster spring.cloud.stream.binders.kafka3.type: kstream -spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster +spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster -spring.cloud.function.definition=process;kstreamProcess +spring.cloud.function.definition: "process;kstreamProcess" # From cluster 1 to cluster 2 with regular process function -spring.cloud.stream.bindings.process-in-0.destination=foo -spring.cloud.stream.bindings.process-in-0.binder=kafka1 # source from cluster 1 -spring.cloud.stream.bindings.process-out-0.destination=bar -spring.cloud.stream.bindings.process-out-0.binder=kafka2 # send to cluster 2 +spring.cloud.stream.bindings.process-in-0.destination: foo +spring.cloud.stream.bindings.process-in-0.binder: kafka1 # source from cluster 1 +spring.cloud.stream.bindings.process-out-0.destination: bar +spring.cloud.stream.bindings.process-out-0.binder: kafka2 # send to cluster 2 # Kafka Streams processor on cluster 2 -spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar -spring.cloud.stream.bindings.kstreamProcess-in-0.binder=kafka3 -spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar -spring.cloud.stream.bindings.kstreamProcess-out-0.binder=kafka3 -``` +spring.cloud.stream.bindings.kstreamProcess-in-0.destination: bar +spring.cloud.stream.bindings.kstreamProcess-in-0.binder: kafka3 +spring.cloud.stream.bindings.kstreamProcess-out-0.destination: foobar +spring.cloud.stream.bindings.kstreamProcess-out-0.binder: kafka3 +---- Pay attention to the above configuration. We have two kinds of binders, but 3 binders all in all, first one is the regular Kafka binder based on cluster 1 (`kafka1`), then another Kafka binder based on cluster 2 (`kafka2`) and finally the `kstream` one (`kafka3`). @@ -68,7 +71,8 @@ Since there are three different binder types available in the Kafka Streams fami For e.g. if you have a processor as below, -``` +[source,java] +---- @Bean public Function, Function, @@ -76,26 +80,27 @@ public Function, ... } -``` +---- then, this has to be configured in a multi binder scenario as the following. Please note that this is only needed if you have a true multi-binder scenario where there are multiple processors dealing with multiple clusters within a single application. In that case, the binders need to be explicitly provided with the bindings to distinguish from other processor's binder types and clusters. -``` +[source,yaml] +---- spring.cloud.stream.binders.kafka1.type: kstream -spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} +spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-2} spring.cloud.stream.binders.kafka2.type: ktable -spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} +spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-2} spring.cloud.stream.binders.kafka3.type: globalktable -spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} +spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers: ${kafkaCluster-2} -spring.cloud.stream.bindings.enrichOrder-in-0.binder=kafka1 #kstream -spring.cloud.stream.bindings.enrichOrder-in-1.binder=kafka2 #ktablr -spring.cloud.stream.bindings.enrichOrder-in-2.binder=kafka3 #globalktable -spring.cloud.stream.bindings.enrichOrder-out-0.binder=kafka1 #kstream +spring.cloud.stream.bindings.enrichOrder-in-0.binder: kafka1 #kstream +spring.cloud.stream.bindings.enrichOrder-in-1.binder: kafka2 #ktablr +spring.cloud.stream.bindings.enrichOrder-in-2.binder: kafka3 #globalktable +spring.cloud.stream.bindings.enrichOrder-out-0.binder: kafka1 #kstream # rest of the configuration is omitted. -``` +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc index 8b3d50aeb3..eb5466a0a2 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc @@ -8,25 +8,28 @@ Let's see some examples. This is the same processor we already saw multiple times, -``` +[source,java] +---- @Bean public Function, KStream> process() { ... } -``` +---- Here is the output binding destination: -``` +[source,yaml] +---- spring.cloud.stream.bindings.process-out-0.destination: outputTopic -``` +---- If the topic `outputTopic` has 4 partitions, if you don't provide a partitioning strategy, Kafka Streams will use default partitioning strategy which may not be the outcome you want depending on the particular use case. Let's say, you want to send any key that matches to `spring` to partition 0, `cloud` to partition 1, `stream` to partition 2, and everything else to partition 3. This is what you need to do in the application. -``` +[source,java] +---- @Bean public StreamPartitioner streamPartitioner() { return (t, k, v, n) -> { @@ -44,16 +47,17 @@ public StreamPartitioner streamPartitioner() { } }; } -``` +---- This is a rudimentary implementation, however, you have access to the key/value of the record, the topic name and the total number of partitions. Therefore, you can implement complex partitioning strategies if need be. You also need to provide this bean name along with the application configuration. -``` +[source,yaml] +---- spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitionerBeanName: streamPartitioner -``` +---- Each output topic in the application needs to be configured separately like this. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc index 5136b511a7..e040e107e2 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc @@ -48,7 +48,8 @@ Once built as a uber-jar (e.g., `kstream-consumer-app.jar`), you can run the abo If the applications choose to define the functional beans using Spring's `Component` annotation, the binder also supports that model. The above functional bean could be rewritten as below. -``` +[source,java] +---- @Component(name = "process") public class SimpleConsumer implements java.util.function.Consumer> { @@ -59,9 +60,9 @@ public class SimpleConsumer implements java.util.function.Consumer, KTable, KStream> process() { @@ -164,11 +165,11 @@ In this example, the first parameter of `BiFunction` is bound as a `KStream` for If there are two inputs, but no outputs, in that case we can use `java.util.function.BiConsumer` as shown below. -[source] +[source,java] ---- @Bean public BiConsumer, KTable> process() { - return (userClicksStream, userRegionsTable) -> {} + return (userClicksStream, userRegionsTable) -> {}; } ---- @@ -183,7 +184,7 @@ Spring Cloud Stream Kafka Streams binder can make use of this feature to enable Let's see an example. -[source] +[source,java] ---- @Bean public Function, @@ -235,18 +236,18 @@ If the application provides a `KTable` as output in the function, the binder sti For example both functions below will work: -``` +[source,java] +---- @Bean public Function, KTable> foo() { return KStream::toTable; - }; } @Bean public Function, KStream> bar() { return KTable::toStream; } -``` +---- [[multiple-output-bindings]] === Multiple Output Bindings @@ -256,7 +257,7 @@ When using multiple output bindings, you need to provide an array of KStream (`K Here is an example: -[source] +[source,java] ---- @Bean public Function, KStream[]> process() { @@ -315,7 +316,8 @@ Kafka Streams binder supports minimal forms of functional composition for linear Using the Java functional API support, you can write multiple functions and then compose them on your own using the `andThen` method. For example, assume that you have the following two functions. -``` +[source,java] +---- @Bean public Function, KStream> foo() { return input -> input.peek((s, s2) -> {}); @@ -325,31 +327,34 @@ public Function, KStream> foo() { public Function, KStream> bar() { return input -> input.peek((s, s2) -> {}); } -``` +---- Even without the functional composition support in the binder, you can compose these two functions as below. -``` +[source,java] +---- @Bean public Function, KStream> composed() { foo().andThen(bar()); } -``` +---- Then you can provide definitions of the form `spring.cloud.function.definition=foo;bar;composed`. With the functional composition support in the binder, you don't need to write this third function in which you are doing explicit function composition. You can simply do this instead: -``` +[source,properties] +---- spring.cloud.function.definition=foo|bar -``` +---- You can even do this: -``` +[source,properties] +---- 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`. @@ -370,7 +375,8 @@ You cannot also compose with functions where the output is an array (`KStream[]` The very first `Function` of `BiFunction` in the function definition may use a curried form also. For example, the following is possible. -``` +[source,java] +---- @Bean public Function, Function, KStream>> curriedFoo() { return a -> b -> @@ -381,7 +387,7 @@ public Function, Function, KStrea public Function, KStream> bar() { return input -> input.mapValues(value -> value + "From-anotherFooFunc"); } -``` +---- and the function definition could be `curriedFoo|bar`. Behind the scenes, the binder will create two input bindings for the curried function, and an output binding based on the final function in the definition. @@ -393,18 +399,18 @@ The default output binding for this example becomes `curriedFoobar-out-0`. Lets say you have the following two functions. -``` +[source,java] +---- @Bean public Function, KTable> foo() { return KStream::toTable; - }; } @Bean public Function, KStream> bar() { return KTable::toStream; } -``` +---- 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. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc index 4083957f47..a9ad6fb979 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc @@ -8,10 +8,12 @@ More often, these failures may self resolve if you can try them again. By default, Kafka Streams binder creates `RetryTemplate` beans for all the input bindings. If the function has the following signature, -``` + +[source,java] +---- @Bean public java.util.function.Consumer> process() -``` +---- and with default binding name, the `RetryTemplate` will be registered as `process-in-0-RetryTemplate`. This is following the convention of binding name (`process-in-0`) followed by the literal `-RetryTemplate`. In the case of multiple input bindings, there will be a separate `RetryTemplate` bean available per binding. @@ -20,7 +22,8 @@ If there is a custom `RetryTemplate` bean available in the application and provi Once the `RetryTemplate` from the binding is injected into the application, it can be used to retry any critical sections of the application. Here is an example: -``` +[source,java] +---- @Bean public java.util.function.Consumer> process(@Lazy @Qualifier("process-in-0-RetryTemplate") RetryTemplate retryTemplate) { @@ -42,11 +45,12 @@ public java.util.function.Consumer> process(@Lazy @Quali } }); } -``` +---- Or you can use a custom `RetryTemplate` as below. -``` +[source,java] +---- @EnableAutoConfiguration public static class CustomRetryTemplateApp { @@ -88,18 +92,20 @@ public static class CustomRetryTemplateApp { }); } } -``` +---- Note that when retries are exhausted, by default, the last exception will be thrown, causing the processor to terminate. If you wish to handle the exception and continue processing, you can add a RecoveryCallback to the `execute` method: Here is an example. -``` + +[source,java] +---- retryTemplate.execute(context -> { //Critical business logic goes here. }, context -> { //Recovery logic goes here. return null; )); -``` +---- 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. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc index 41abba4a67..114517e210 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc @@ -9,14 +9,15 @@ Both of these customizers are part of the Spring for Apache Kafka project. Here is an example of using the `StreamsBuilderFactoryBeanConfigurer`. -``` +[source,java] +---- @Bean public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() { return sfb -> sfb.setStateListener((newState, oldState) -> { //Do some action here! }); } -``` +---- The above is shown as an illustration of the things you can do to customize the `StreamsBuilderFactoryBean`. You can essentially call any available mutation operations from `StreamsBuilderFactoryBean` to customize it. @@ -25,7 +26,8 @@ This customizer will be invoked by the binder right before the factory bean is s Once you get access to the `StreamsBuilderFactoryBean`, you can also customize the underlying `KafkaStreams` object. Here is a blueprint for doing so. -``` +[source,java] +---- @Bean public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() { return factoryBean -> { @@ -39,7 +41,7 @@ public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() }); }; } -``` +---- `KafkaStreamsCustomizer` will be called by the `StreamsBuilderFactoryBeabn` right before the underlying `KafkaStreams` gets started. @@ -48,7 +50,8 @@ If the customization needs to be different for multiple Kafka Streams processors For e.g, -``` +[source,java] +---- @Bean public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() { return factoryBean -> { @@ -64,7 +67,8 @@ public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() }); } }; -``` +} +---- [[using-configurer-to-register-a-global-state-store]] == Using StreamsBuilderFactoryBeanConfigurer to register a global state store @@ -73,7 +77,8 @@ As mentioned above, the binder does not provide a first class way to register gl For that, you need to use the customizer via `StreamsBuilderFactoryBeanConfigurer`. Here is how that can be done. -``` +[source,java] +---- @Bean public StreamsBuilderFactoryBeanConfigurer customizer() { return streamsBuilderFactoryBean -> { @@ -92,7 +97,7 @@ public StreamsBuilderFactoryBeanConfigurer customizer() { } }; } -``` +---- Any customizations on `StreamsBuilder` must be done through the `KafkaStreamsInfrastructureCustomizer` as shown above. If `StreamsBuilderFactoryBean#getObject()` is called to get access to the `StreamsBuilder` object, that may not work as the bean maybe in initialization and thus run into some circular dependency issues. @@ -105,7 +110,8 @@ If you have multiple processors, you want to attach the global state store to th 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 `StreamsBuilderFactoryBean` customizer to register production exception handlers. See below. -``` +[source,java] +---- @Bean public StreamsBuilderFactoryBeanConfigurer configurer() { return fb -> { @@ -113,7 +119,7 @@ public StreamsBuilderFactoryBeanConfigurer configurer() { CustomProductionExceptionHandler.class); }; } -``` +---- 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. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc index 8372ddcceb..13b78ec697 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc @@ -6,7 +6,8 @@ By default, Kafka Streams extracts the timestamp metadata embedded in the consum You can change this default behavior by providing a different `TimestampExtractor` implementation per input binding. Here are some details on how that can be done. -``` +[source,java] +---- @Bean public Function, Function, @@ -20,15 +21,16 @@ public Function, public TimestampExtractor timestampExtractor() { return new WallclockTimestampExtractor(); } -``` +---- Then you set the above `TimestampExtractor` bean name per consumer binding. -``` +[source,properties] +---- spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.timestampExtractorBeanName=timestampExtractor spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.timestampExtractorBeanName=timestampExtractor -spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.timestampExtractorBeanName=timestampExtractor" -``` +spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.timestampExtractorBeanName=timestampExtractor +---- If you skip an input consumer binding for setting a custom timestamp extractor, that consumer will use the default settings. diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc index 78f23c2677..c233ffe384 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc @@ -6,7 +6,8 @@ However, in order to trace any application specific operations, those need to be This can be done by injecting the `KafkaStreamsTracing` bean from Spring Cloud Sleuth in the application and then invoke various Kafka Streams operations through this injected bean. Here are some examples of using it. -``` +[source,java] +---- @Bean public BiFunction, KTable, KStream> clicks(KafkaStreamsTracing kafkaStreamsTracing) { return (userClicksStream, userRegionsTable) -> (userClicksStream @@ -23,7 +24,7 @@ public BiFunction, KTable, KStream, KStream> process(KafkaStreamsTracing kafkaStreamsTracing) { return input -> input.transform(kafkaStreamsTracing.transformer( @@ -60,5 +62,5 @@ public Function, KStream> process(KafkaS } })); } -``` +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index 67a3ddf9d0..324ed20685 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc @@ -762,7 +762,8 @@ refer to the <> for mo You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application. -``` +[source,java] +---- @Component class NoOpBindingMeters { NoOpBindingMeters(MeterRegistry registry) { @@ -770,7 +771,7 @@ class NoOpBindingMeters { MeterFilter.denyNameStartsWith(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)); } } -``` +---- More details on how to suppress meters selectively can be found https://micrometer.io/docs/concepts#_meter_filters[here]. diff --git a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc index f32ed35d1f..8476fca3e9 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc @@ -17,32 +17,34 @@ Instead, I want to send the record in error to a DLT (Dead-Letter-Topic) and the The solution for this problem is to use the DLQ feature in Spring Cloud Stream. For the purposes of this discussion, let us assume that the following is our processor function. -``` +[source,java] +---- @Bean public Consumer processData() { return s -> { throw new RuntimeException(); }; } -``` +---- This is a very trivial function that throws an exception for all the records that it processes, but you can take this function and extend it to any other similar situations. In order to send the records in error to a DLT, we need to provide the following configuration. -``` +[source,yaml] +---- spring.cloud.stream: bindings: processData-in-0: group: my-group destination: input-topic - kafka: + kafka: bindings: processData-in-0: consumer: enableDlq: true dlqName: input-topic-dlq -``` +---- In order to activate DLQ, the application must provide a group name. Anonymous consumers cannot use the DLQ facilities. @@ -54,25 +56,27 @@ By default, the DLQ producer in Kafka binder expects the payload of type `byte[] If that is not the case, then we need to provide the configuration for proper serializer. For example, let us re-write the consumer function as below: -``` +[source,java] +---- @Bean public Consumer processData() { return s -> { throw new RuntimeException(); }; } -``` +---- Now, we need to tell Spring Cloud Stream, how we want to serialize the data when writing to the DLT. Here is the modified configuration for this scenario: -``` +[source,yaml] +---- spring.cloud.stream: bindings: processData-in-0: group: my-group destination: input-topic - kafka: + kafka: bindings: processData-in-0: consumer: @@ -82,7 +86,7 @@ spring.cloud.stream: configuration: value.serializer: org.apache.kafka.common.serialization.StringSerializer -``` +---- [[dlq-with-advanced-retry-options]] == DLQ with Advanced Retry Options @@ -100,36 +104,40 @@ If you followed the above recipe, then you get the default retry options built i By default, the binder retires for a maximum of 3 attempts with a one second initial delay, 2.0 multiplier with each back off with a max delay of 10 seconds. You can change all these configurations as below: -``` +[source,properties] +---- spring.cloud.stream.bindings.processData-in-0.consumer.maxAttempts spring.cloud.stream.bindings.processData-in-0.consumer.backOffInitialInterval spring.cloud.stream.bindings.processData-in-0.consumer.backOffMultiplier spring.cloud.stream.bindings.processData-in-0.consumer.backOffMaxInterval -``` +---- If you want, you can also provide a list of retryable exceptions by providing a map of boolean values. For example, -``` +[source,properties] +---- spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalStateException=true spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalArgumentException=false -``` +---- By default, any exceptions not listed in the map above will be retried. If that is not desired, then you can disable that by providing, -``` +[source,properties] +---- spring.cloud.stream.bindings.processData-in-0.consumer.defaultRetryable=false -``` +---- You can also provide your own `RetryTemplate` and mark it as `@StreamRetryTemplate` which will be scanned and used by the binder. This is useful when you want more sophisticated retry strategies and policies. If you have multiple `@StreamRetryTemplate` beans, then you can specify which one your binding wants by using the property, -``` +[source,properties] +---- spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name= -``` +---- [[handling-deserialization-errors-with-dlq]] == Handling Deserialization errors with DLQ @@ -151,14 +159,15 @@ Let us explore those. Assuming this is our function: -``` +[source,java] +---- @Bean public Consumer functionName() { return s -> { System.out.println(s); }; } -``` +---- It is a trivial function that takes a `String` parameter. @@ -173,28 +182,31 @@ This `DefaultErrorHandler` is configured with a `DeadLetterPublishingRecoverer`. We also need to configure an `ErrorHandlingDeserializer` for the consumer. That sounds like a lot of complex things, but in reality, it boils down to these 3 beans in this case. -``` +[source,java] +---- @Bean public ListenerContainerCustomizer> customizer(DefaultErrorHandler errorHandler) { return (container, dest, group) -> { container.setCommonErrorHandler(errorHandler); }; } -``` +---- -``` +[source,java] +---- @Bean public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) { return new DefaultErrorHandler(deadLetterPublishingRecoverer); } -``` +---- -``` +[source,java] +---- @Bean public DeadLetterPublishingRecoverer publisher(KafkaOperations bytesTemplate) { return new DeadLetterPublishingRecoverer(bytesTemplate); } -``` +---- Let us analyze each of them. The first one is the `ListenerContainerCustomizer` bean that takes a `DefaultErrorHandler`. @@ -218,7 +230,8 @@ It then sets the exception in a header and passes this record to the listener, w Following is the configuration required: -``` +[source,yaml] +---- spring.cloud.stream: function: definition: functionName @@ -240,7 +253,7 @@ spring.cloud.stream: configuration: value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer -``` +---- We are providing the `ErrorHandlingDeserializer` through the `configuration` property on the binding. We are also indicating that the actual deserializer to delegate is the `StringDeserializer`. @@ -305,7 +318,8 @@ First of all, when you want to reset to an arbitrary offset other than `earliest Then you have to provide a custom bean of type `KafkaBindingRebalanceListener`, which will be injected into all consumer bindings. It is an interface that comes with a few default methods, but here is the method that we are interested in: -``` +[source,java] +---- /** * Invoked when partitions are initially assigned or after a rebalance. Applications * might only want to perform seek operations on an initial assignment. While the @@ -323,7 +337,7 @@ It is an interface that comes with a few default methods, but here is the method Collection partitions, boolean initial) { // do nothing } -``` +---- Let us look at the details. @@ -334,18 +348,20 @@ When the consumer starts for the first time, all 4 partitions are getting initia However, we do not want to start the partitions to consume at the defaults (`earliest` since we define a group), rather for each partition, we want them to consume after seeking to arbitrary offsets. Imagine that you have a business case to consume from certain offsets as below. -``` +[source] +---- Partition start offset 0 1000 1 2000 2 2000 3 1000 -``` +---- This could be achieved by implementing the above method as below. -``` +[source,java] +---- @Override public void onPartitionsAssigned(String bindingName, Consumer consumer, Collection partitions, boolean initial) { @@ -370,13 +386,13 @@ public void onPartitionsAssigned(String bindingName, Consumer consumer, Co }); } } -``` +---- This is just a rudimentary implementation. Real world use cases are much more complex than this and you need to adjust accordingly, but this certainly gives you a basic sketch. When consumer `seek` fails, it may throw some runtime exceptions and you need to decide what to do in those cases. -[[what-if-we-start-a-second-consumer-with-the-same-group-id?]] +[[what-if-we-start-a-second-consumer-with-the-same-group-id]] === What if we start a second consumer with the same group id? When we add a second consumer, a rebalance will occur and some partitions will be moved around. @@ -387,7 +403,7 @@ In the case of the first consumer, it now only has partitions `0` and `1` However, for this consumer it was simply a rebalance event and not considered as an initial assignment. Thus, it will not re-seek to the given offsets because of the conditional check on the `initial` argument. -[[how-do-i-manually-acknowledge-using-kafka-binder?]] +[[how-do-i-manually-acknowledge-using-kafka-binder]] == How do I manually acknowledge using Kafka binder? [[manually-acknowledging-problem-statement]] @@ -411,7 +427,8 @@ When it is set like that, then there will be a header called `kafka_acknowledgme For example, imagine this as your consumer method. -``` +[source,java] +---- @Bean public Consumer> myConsumer() { return msg -> { @@ -422,11 +439,11 @@ public Consumer> myConsumer() { } }; } -``` +---- Then you set the property `spring.cloud.stream.kafka.bindings.myConsumer-in-0.consumer.ackMode` to `MANUAL` or `MANUAL_IMMEDIATE`. -[[how-do-i-override-the-default-binding-names-in-spring-cloud-stream?]] +[[how-do-i-override-the-default-binding-names-in-spring-cloud-stream]] == How do I override the default binding names in Spring Cloud Stream? [[override-the-default-binding-names-problem-statement]] @@ -439,12 +456,13 @@ Spring Cloud Stream creates default bindings based on the function definition an Assume that following is your function signature. -``` +[source,java] +---- @Bean public Function uppercase(){ ... } -``` +---- By default, Spring Cloud Stream will create the bindings as below. @@ -453,21 +471,23 @@ By default, Spring Cloud Stream will create the bindings as below. You can override these bindings to something by using the following properties. -``` +[source,properties] +---- spring.cloud.stream.function.bindings.uppercase-in-0=my-transformer-in spring.cloud.stream.function.bindings.uppercase-out-0=my-transformer-out -``` +---- After this, all binding properties must be made on the new names, `my-transformer-in` and `my-transformer-out`. Here is another example with Kafka Streams and multiple inputs. -``` +[source,java] +---- @Bean public BiFunction, KTable, KStream> processOrder() { ... } -``` +---- By default, Spring Cloud Stream will create three different binding names for this function. @@ -490,7 +510,7 @@ You can easily do that by simply setting these three properties Once you do that, it overrides the default binding names and any properties that you want to set on them must be on these new binding names. -[[how-do-i-send-a-message-key-as-part-of-my-record?]] +[[how-do-i-send-a-message-key-as-part-of-my-record]] == How do I send a message key as part of my record? [[send-a-message-key-problem-statement]] @@ -507,27 +527,29 @@ Following is a basic blueprint for doing this, but you may want to adapt it to y Here is sample producer method (aka `Supplier`). -``` +[source,java] +---- @Bean public Supplier> supplier() { return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build(); } -``` +---- This is a trivial function that sends a message with a `String` payload, but also with a key. Note that we set the key as a message header using `KafkaHeaders.MESSAGE_KEY`. If you want to change the key from the default `kafka_messageKey`, then in the configuration, we need to specify this property: -``` +[source,properties] +---- spring.cloud.stream.kafka.bindings.supplier-out-0.producer.messageKeyExpression=headers['my-special-key'] -``` +---- Please note that we use the binding name `supplier-out-0` since that is our function name, please update accordingly. Then, we use this new key when we produce the message. -[[how-do-i-use-native-serializer-and-deserializer-instead-of-message-conversion-done-by-spring-cloud-stream?]] +[[how-do-i-use-native-serializer-and-deserializer-instead-of-message-conversion-done-by-spring-cloud-stream]] == How do I use native serializer and deserializer instead of message conversion done by Spring Cloud Stream? [[use-native-serializer-and-deserializer-problem-statement]] @@ -544,24 +566,27 @@ This is really easy to do. All you have to do is to provide the following property to enable native serialization. -``` +[source,yaml] +---- spring.cloud.stream.kafka.bindings..producer.useNativeEncoding: true -``` +---- Then, you need to also set the serializers. There are a couple of ways to do this. -``` +[source,yaml] +---- spring.cloud.stream.kafka.bindings..producer.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer spring.cloud.stream.kafka.bindings..producer.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer -``` +---- or using the binder configuration. -``` +[source,yaml] +---- spring.cloud.stream.kafka.binder.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer spring.cloud.stream.kafka.binder.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer -``` +---- When using the binder way, it is applied against all the bindings whereas setting them at the bindings are per binding. @@ -569,18 +594,20 @@ On the deserializing side, you just need to provide the deserializers as configu For example, -``` +[source,yaml] +---- spring.cloud.stream.kafka.bindings..consumer.configuration.key.deserializer: org.apache.kafka.common.serialization.StringDeserializer spring.cloud.stream.kafka.bindings..producer.configuration.value.deserializer: org.apache.kafka.common.serialization.StringDeserializer -``` +---- You can also set them at the binder level. There is an optional property that you can set to force native decoding. -``` +[source,yaml] +---- spring.cloud.stream.kafka.bindings..consumer.useNativeDecoding: true -``` +---- However, in the case of Kafka binder, this is unnecessary, as by the time it reaches the binder, Kafka already deserializes them using the configured deserializers. @@ -599,13 +626,14 @@ Kafka Streams binder allows you to do that. Before we look at the solution, let us look at the following scenario. -``` +[source,java] +---- @Bean public BiConsumer, KTable> myBiConsumer{ (s, t) -> s.join(t, ...) ... } -``` +---- We have a `BiConsumer` bean that requires two input bindings. In this case, the first binding is for a `KStream` and the second one is for a `KTable`. @@ -613,10 +641,11 @@ When running this application for the first time, by default, both bindings star What about I want to start from the `latest` offset due to some requirements? You can do this by enabling the following properties. -``` +[source,yaml] +---- spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-0.consumer.startOffset: latest spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-1.consumer.startOffset: latest -``` +---- If you want only one binding to start from the `latest` offset and the other to consumer from the default `earliest`, then leave the latter binding out from the configuration. @@ -635,39 +664,43 @@ I have a Kafka producer application and I want to keep track of all my successfu Let us assume that we have this following supplier in the application. -``` +[source,java] +---- @Bean public Supplier> supplier() { return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build(); } -``` +---- Then, we need to define a new `MessageChannel` bean to capture all the successful send information. -``` +[source,java] +---- @Bean public MessageChannel fooRecordChannel() { return new DirectChannel(); } -``` +---- Next, define this property in the application configuration to provide the bean name for the `recordMetadataChannel`. -``` +[source,yaml] +---- spring.cloud.stream.kafka.bindings.supplier-out-0.producer.recordMetadataChannel: fooRecordChannel -``` +---- At this point, successful sent information will be sent to the `fooRecordChannel`. You can write an `IntegrationFlow` as below to see the information. -``` +[source,java] +---- @Bean public IntegrationFlow integrationFlow() { return f -> f.channel("fooRecordChannel") .handle((payload, messageHeaders) -> payload); } -``` +---- In the `handle` method, the payload is what got sent to Kafka and the message headers contain a special key called `kafka_recordMetadata`. Its value is a `RecordMetadata` that contains information about topic partition, current offset etc. @@ -687,16 +720,18 @@ Under normal circumstances, this should be fine. Imagine, you have the following producer. -``` +[source,java] +---- @Bean public Supplier> supply() { return () -> MessageBuilder.withPayload("foo").setHeader("foo", "bar").build(); } -``` +---- On the consumer side, you should still see the header "foo", and the following should not give you any issues. -``` +[source,java] +---- @Bean public Consumer> consume() { return s -> { @@ -704,12 +739,13 @@ public Consumer> consume() { System.out.println(foo); }; } -``` +---- If you provide a https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.3/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_binder_properties[custom header mapper] in the application, then this won't work. Let's say you have an empty `KafkaHeaderMapper` in the application. -``` +[source,java] +---- @Bean public KafkaHeaderMapper kafkaBinderHeaderMapper() { return new KafkaHeaderMapper() { @@ -724,13 +760,14 @@ public KafkaHeaderMapper kafkaBinderHeaderMapper() { } }; } -``` +---- If that is your implementation, then you will miss the `foo` header on the consumer. Chances are that, you may have some logic inside those `KafkaHeaderMapper` methods. You need the following to populate the `foo` header. -``` +[source,java] +---- @Bean public KafkaHeaderMapper kafkaBinderHeaderMapper() { return new KafkaHeaderMapper() { @@ -743,10 +780,11 @@ public KafkaHeaderMapper kafkaBinderHeaderMapper() { @Override public void toHeaders(Headers source, Map target) { final Header foo = source.lastHeader("foo"); - target.put("foo", new String(foo.value())); + target.put("foo", new String(foo.value())); } - } -``` + }; +} +---- That will properly populate the `foo` header from the producer to consumer. @@ -760,17 +798,19 @@ See this https://stackoverflow.com/questions/68412600/change-the-behaviour-in-sp In that case, you must use a custom `KafkaHeaderMapper` to map the case-sensitive id header. For example, let's say you have the following producer. -``` +[source,java] +---- @Bean public Supplier> supply() { return () -> MessageBuilder.withPayload("foo").setHeader("Id", "my-id").build(); } -``` +---- The header `Id` above will be gone from the consuming side as it clashes with the framework `id` header. You can provide a custom `KafkaHeaderMapper` to solve this issue. -``` +[source,java] +---- @Bean public KafkaHeaderMapper kafkaBinderHeaderMapper1() { return new KafkaHeaderMapper() { @@ -787,7 +827,7 @@ public KafkaHeaderMapper kafkaBinderHeaderMapper1() { } }; } -``` +---- By doing this, both `id` and `Id` headers will be available from the producer to the consumer side. @@ -809,7 +849,8 @@ In order to produce to multiple topics, use `StreamBridge` API. Below are the code snippets for this: -``` +[source,java] +---- @Autowired StreamBridge bridge; @@ -841,13 +882,14 @@ DefaultAfterRollbackProcessor rollbackProcessor(KafkaTemplate te new DeadLetterPublishingRecoverer(template), new FixedBackOff(2000L, 2L), template, true); } -``` +---- [[required-configuration]] === Required Configuration -``` -spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: tx- +[source,properties] +---- +spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx- spring.cloud.stream.kafka.binder.required-acks=all spring.cloud.stream.bindings.input-in-0.group=foo spring.cloud.stream.bindings.input-in-0.destination=input @@ -855,11 +897,12 @@ spring.cloud.stream.bindings.left.destination=left spring.cloud.stream.bindings.right.destination=right spring.cloud.stream.kafka.bindings.input-in-0.consumer.maxAttempts=1 -``` +---- in order to test, you can use the following: -``` +[source,java] +---- @Bean public ApplicationRunner runner(KafkaTemplate template) { return args -> { @@ -868,7 +911,7 @@ public ApplicationRunner runner(KafkaTemplate template) { template.send("input", "Good".getBytes()); }; } -``` +---- Some important notes: @@ -894,18 +937,20 @@ How can I run multiple instances of the pollable consumers and generate unique ` Assuming that I have the following definition: -``` +[source,yaml] +---- spring.cloud.stream.pollable-source: foo spring.cloud.stream.bindings.foo-in-0.group: my-group -``` +---- When running the application, the Kafka consumer generates a client.id (something like `consumer-my-group-1`). For each instance of the application that is running, this `client.id` will be the same, causing unexpected issues. In order to fix this, you can add the following property on each instance of the application: -``` +[source,properties] +---- spring.cloud.stream.kafka.bindings.foo-in-0.consumer.configuration.client.id=${client.id} -``` +---- See this https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1139[GitHub issue] for more details. diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_overview/health-indicator.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_overview/health-indicator.adoc index 2b64c62747..51a66a20da 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_overview/health-indicator.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_overview/health-indicator.adoc @@ -9,10 +9,11 @@ In the case of multi-binder environments, this has to be set on the binder's env When the health indicator is disabled, you should see something like the below in the health actuator endpoint: -``` +[source,json] +---- "rabbit": { "status": "UNKNOWN" } -``` +---- At the Spring Boot level, if you want to disable the Rabbit health indicator, you need to use the property `management.health.rabbit.enabled` and set to `false`. diff --git a/tools/kafka/docker-compose/README.adoc b/tools/kafka/docker-compose/README.adoc index 09899feacc..751bedf27f 100644 --- a/tools/kafka/docker-compose/README.adoc +++ b/tools/kafka/docker-compose/README.adoc @@ -25,18 +25,20 @@ Now that we started our clusters, let us try to produce data to a topic on the c To produce data to a topic named `input-topic`: -``` +[source,shell] +---- docker exec -it broker1 /bin/kafka-console-producer --bootstrap-server localhost:29091 --topic input-topic -``` +---- Note the use of bootstrap-server at `localhost:29091`. This port is internal to the docker container. To consume from the `input-topic`: -``` +[source,shell] +---- docker exec -it broker1 /bin/kafka-console-consumer --bootstrap-server localhost:29091 --topic input-topic -``` +---- Enter some data on the producer console and hit enter, you should see the same data gets echoed on the consumer console.