diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index d3157662c..fd768bd5a 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -40,9 +40,61 @@ * Binders ** Apache Kafka *** Kafka Binder -**** xref:kafka/kafka_dlq.adoc[] -*** Kafka Stream Binder +**** xref:kafka/kafka-binder/usage.adoc[] +**** xref:kafka/kafka-binder/overview.adoc[] +**** xref:kafka/kafka-binder/config-options.adoc[] +**** xref:kafka/kafka-binder/reset-offsets.adoc[] +**** xref:kafka/kafka-binder/consume-batches.adoc[] +**** xref:kafka/kafka-binder/manual-ack.adoc[] +**** xref:kafka/kafka-binder/security-config.adoc[] +**** xref:kafka/kafka-binder/pause_resume.adoc[] +**** xref:kafka/kafka-binder/transactional.adoc[] +**** xref:kafka/kafka-binder/error-channels.adoc[] +**** xref:kafka/kafka-binder/metrics.adoc[] +**** xref:kafka/kafka-binder/tombstone.adoc[] +**** xref:kafka/kafka-binder/rebalance_listener.adoc[] +**** xref:kafka/kafka-binder/retry-dlq.adoc[] +**** xref:kafka/kafka-binder/cons-prod-config-cust.adoc[] +**** xref:kafka/kafka-binder/admin-client-config-cust.adoc[] +**** xref:kafka/kafka-binder/custom-health-ind.adoc[] +**** xref:kafka/kafka-binder/dlq.adoc[] +**** xref:kafka/kafka-binder/partitions.adoc[] *** Reactive Kafka Binder +**** xref:kafka/kafka-reactive-binder/overview.adoc[] +**** xref:kafka/kafka-reactive-binder/usage.adoc[] +**** xref:kafka/kafka-reactive-binder/examples.adoc[] +**** xref:kafka/kafka-reactive-binder/consuming.adoc[] +**** xref:kafka/kafka-reactive-binder/concurrency.adoc[] +**** xref:kafka/kafka-reactive-binder/multiplex.adoc[] +**** xref:kafka/kafka-reactive-binder/pattern.adoc[] +**** xref:kafka/kafka-reactive-binder/sender_result.adoc[] +*** Kafka Stream Binder +**** xref:kafka/kafka-streams-binder/usage.adoc[] +**** xref:kafka/kafka-streams-binder/overview.adoc[] +**** xref:kafka/kafka-streams-binder/programming-model.adoc[] +**** xref:kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc[] +**** xref:kafka/kafka-streams-binder/record-serialization-and-deserialization.adoc[] +**** xref:kafka/kafka-streams-binder/error-handling.adoc[] +**** xref:kafka/kafka-streams-binder/retrying-critical-business-logic.adoc[] +**** xref:kafka/kafka-streams-binder/state-store.adoc[] +**** xref:kafka/kafka-streams-binder/interactive-queries.adoc[] +**** xref:kafka/kafka-streams-binder/health-indicator.adoc[] +**** xref:kafka/kafka-streams-binder/accessing-metrics.adoc[] +**** xref:kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc[] +**** xref:kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc[] +**** xref:kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc[] +**** xref:kafka/kafka-streams-binder/timestamp-extractor.adoc[] +**** xref:kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc[] +**** xref:kafka/kafka-streams-binder/state-cleanup.adoc[] +**** xref:kafka/kafka-streams-binder/topology-visualization.adoc[] +**** xref:kafka/kafka-streams-binder/event-type-based-routing-in-applications.adoc[] +**** xref:kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc[] +**** xref:kafka/kafka-streams-binder/manually-starting-processors.adoc[] +**** xref:kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc[] +**** xref:kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc[] +**** xref:kafka/kafka-streams-binder/configuration-options.adoc[] +*** Tips and Recipes +**** xref:kafka/kafka_tips.adoc[] ** RabbitMQ ** 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] @@ -86,33 +138,33 @@ #* xref:README.adoc[] #* xref:configprops.adoc[] #* xref:intro.adoc[] -#** xref:kafka/kafka-reactive.adoc[] +#** xref:kafka/kafka_reactive_binder.adoc[] #** xref:kafka/kafka-streams.adoc[] -#*** xref:kafka/streams/usage.adoc[] -#*** xref:kafka/streams/overview.adoc[] -#*** xref:kafka/streams/programming-model.adoc[] -#*** xref:kafka/streams/ancillaries-to-the-programming-model.adoc[] -#*** xref:kafka/streams/record-serialization-and-deserialization.adoc[] -#*** xref:kafka/streams/error-handling.adoc[] -#*** xref:kafka/streams/retrying-critical-business-logic.adoc[] -#*** xref:kafka/streams/state-store.adoc[] -#*** xref:kafka/streams/interactive-queries.adoc[] -#*** xref:kafka/streams/health-indicator.adoc[] -#*** xref:kafka/streams/accessing-metrics.adoc[] -#*** xref:kafka/streams/mixing-high-level-dsl-and-low-level-processor-api.adoc[] -#*** xref:kafka/streams/partition-support-on-the-outbound.adoc[] -#*** xref:kafka/streams/streamsbuilderfactorybean-customizer.adoc[] -#*** xref:kafka/streams/timestamp-extractor.adoc[] -#*** xref:kafka/streams/multi-binders-with-based-binders-and-regular-binder.adoc[] -#*** xref:kafka/streams/state-cleanup.adoc[] -#*** xref:kafka/streams/topology-visualization.adoc[] -#*** xref:kafka/streams/event-type-based-routing-in-applications.adoc[] -#*** xref:kafka/streams/binding-visualization-and-control-in-binder.adoc[] -#*** xref:kafka/streams/manually-starting-processors.adoc[] -#*** xref:kafka/streams/manually-starting-processors-selectively.adoc[] -#*** xref:kafka/streams/tracing-using-spring-cloud-sleuth.adoc[] -#*** xref:kafka/streams/configuration-options.adoc[] -#** xref:kafka/kafka_dlq.adoc[] +#*** xref:kafka/kafka-streams-binder/usage.adoc[] +#*** xref:kafka/kafka-streams-binder/overview.adoc[] +#*** xref:kafka/kafka-streams-binder/programming-model.adoc[] +#*** xref:kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc[] +#*** xref:kafka/kafka-streams-binder/record-serialization-and-deserialization.adoc[] +#*** xref:kafka/kafka-streams-binder/error-handling.adoc[] +#*** xref:kafka/kafka-streams-binder/retrying-critical-business-logic.adoc[] +#*** xref:kafka/kafka-streams-binder/state-store.adoc[] +#*** xref:kafka/kafka-streams-binder/interactive-queries.adoc[] +#*** xref:kafka/kafka-streams-binder/health-indicator.adoc[] +#*** xref:kafka/kafka-streams-binder/accessing-metrics.adoc[] +#*** xref:kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc[] +#*** xref:kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc[] +#*** xref:kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc[] +#*** xref:kafka/kafka-streams-binder/timestamp-extractor.adoc[] +#*** xref:kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc[] +#*** xref:kafka/kafka-streams-binder/state-cleanup.adoc[] +#*** xref:kafka/kafka-streams-binder/topology-visualization.adoc[] +#*** xref:kafka/kafka-streams-binder/event-type-based-routing-in-applications.adoc[] +#*** xref:kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc[] +#*** xref:kafka/kafka-streams-binder/manually-starting-processors.adoc[] +#*** xref:kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc[] +#*** xref:kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc[] +#*** xref:kafka/kafka-streams-binder/configuration-options.adoc[] +#** xref:kafka/kafka-binder/dlq.adoc[] #** xref:kafka/kafka_overview.adoc[] #*** xref:kafka/kafka_overview/usage.adoc[] #*** xref:kafka/kafka_overview/overview.adoc[] @@ -127,7 +179,7 @@ #*** xref:kafka/kafka_overview/admin-client-config-customization.adoc[] #*** xref:kafka/kafka_overview/custom-binder-health-indicator.adoc[] #*** xref:kafka/kafka_overview/custom-binder-health-indicator-example.adoc[] -#** xref:kafka/kafka_partitions.adoc[] +#** xref:kafka/kafka-binder/partitions.adoc[] #** xref:kafka/kafka_tips.adoc[] #** xref:kafka/spring-cloud-stream-binder-kafka.adoc[] #** xref:pulsar/pulsar_binder.adoc[] diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/admin-client-config-customization.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc similarity index 96% rename from docs/modules/ROOT/pages/kafka/kafka_overview/admin-client-config-customization.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc index f4fb3c7ba..ef4739392 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/admin-client-config-customization.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/admin-client-config-cust.adoc @@ -1,6 +1,5 @@ [[admin-client-config-customization]] = Customizing AdminClient Configuration -:page-section-summary-toc: 1 As with consumer and producer config customization above, applications can also customize the configuration for admin clients by providing an `AdminClientConfigCustomizer`. AdminClientConfigCustomizer's configure method provides access to the admin client properties, using which you can define further customization. @@ -15,4 +14,3 @@ public AdminClientConfigCustomizer adminClientConfigCustomizer() { }; } ``` - diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/configuration-options.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/config-options.adoc similarity index 66% rename from docs/modules/ROOT/pages/kafka/kafka_overview/configuration-options.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/config-options.adoc index 53d27cb72..1f3552ceb 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/configuration-options.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/config-options.adoc @@ -1,4 +1,3 @@ -[[configuration-options]] = Configuration Options This section contains the configuration options used by the Apache Kafka binder. @@ -88,7 +87,7 @@ When transactions are enabled, individual `producer` properties are ignored and Default `null` (no transactions) spring.cloud.stream.kafka.binder.transaction.producer.*:: Global producer properties for producers in a transactional binder. -See `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` and xref:kafka/kafka_overview/configuration-options.adoc#kafka-producer-properties[Kafka Producer Properties] and the general producer properties supported by all binders. +See `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` and <> and the general producer properties supported by all binders. + Default: See individual producer properties. @@ -102,7 +101,7 @@ Default: none. spring.cloud.stream.kafka.binder.considerDownWhenAnyPartitionHasNoLeader:: Flag to set the binder health as `down`, when any partitions on the topic, regardless of the consumer that is receiving data from it, is found without a leader. + -Default: `false`. +Default: `true`. spring.cloud.stream.kafka.binder.certificateStoreDirectory:: When the truststore or keystore certificate location is given as a non-local file system resource (resources supported by org.springframework.core.io.Resource e.g. CLASSPATH, HTTP, etc.), @@ -195,15 +194,15 @@ Note that this property is only applicable for pollable consumers. Default: not set. resetOffsets:: Whether to reset offsets on the consumer to the value provided by startOffset. -Must be false if a `KafkaBindingRebalanceListener` is provided; see xref:kafka/kafka_overview/rebalance-listener.adoc[Using a KafkaBindingRebalanceListener]. -See xref:kafka/kafka_overview/configuration-options.adoc#reset-offsets[Resetting Offsets] for more information about this property. +Must be false if a `KafkaBindingRebalanceListener` is provided; see <>. +See <> for more information about this property. + Default: `false`. startOffset:: The starting offset for new groups. Allowed values: `earliest` and `latest`. If the consumer group is set explicitly for the consumer 'binding' (through `spring.cloud.stream.bindings..group`), 'startOffset' is set to `earliest`. Otherwise, it is set to `latest` for the `anonymous` consumer group. -See xref:kafka/kafka_overview/configuration-options.adoc#reset-offsets[Resetting Offsets] for more information about this property. +See <> for more information about this property. + Default: null (equivalent to `earliest`). enableDlq:: @@ -211,17 +210,17 @@ When set to true, it enables DLQ behavior for the consumer. By default, messages that result in errors are forwarded to a topic named `error..`. The DLQ topic name can be configurable by setting the `dlqName` property or by defining a `@Bean` of type `DlqDestinationResolver`. This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome. -See xref:kafka/kafka_dlq.adoc[Dead-Letter Topic Processing] processing for more information. +See <> processing for more information. Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: `x-original-topic`, `x-exception-message`, and `x-exception-stacktrace` as `byte[]`. By default, a failed record is sent to the same partition number in the DLQ topic as the original record. -See xref:kafka/kafka_dlq.adoc#dlq-partition-selection[Dead-Letter Topic Partition Selection] for how to change that behavior. +See <> for how to change that behavior. **Not allowed when `destinationIsPattern` is `true`.** + Default: `false`. dlqPartitions:: When `enableDlq` is true, and this property is not set, a dead letter topic with the same number of partitions as the primary topic(s) is created. Usually, dead-letter records are sent to the same partition in the dead-letter topic as the original record. -This behavior can be changed; see xref:kafka/kafka_dlq.adoc#dlq-partition-selection[Dead-Letter Topic Partition Selection]. +This behavior can be changed; see <>. If this property is set to `1` and there is no `DqlPartitionFunction` bean, all dead-letter records will be written to partition `0`. If this property is greater than `1`, you **MUST** provide a `DlqPartitionFunction` bean. Note that the actual partition count is affected by the binder's `minPartitionCount` property. @@ -258,7 +257,7 @@ Default: `null` idleEventInterval:: The interval, in milliseconds, between events indicating that no messages have recently been received. Use an `ApplicationListener` to receive these events. -See xref:kafka/kafka_overview/configuration-options.adoc#pause-resume[Example: Pausing and Resuming the Consumer] for a usage example. +See <> for a usage example. + Default: `30000` destinationIsPattern:: @@ -305,49 +304,6 @@ This is a handy way to express error handlers, if the application does not want + Default: none. - -[[reset-offsets]] -== Resetting Offsets - -When an application starts, the initial position in each assigned partition depends on two properties `startOffset` and `resetOffsets`. -If `resetOffsets` is `false`, normal Kafka consumer https://kafka.apache.org/documentation/#consumerconfigs_auto.offset.reset[`auto.offset.reset`] semantics apply. -i.e. If there is no committed offset for a partition for the binding's consumer group, the position is `earliest` or `latest`. -By default, bindings with an explicit `group` use `earliest`, and anonymous bindings (with no `group`) use `latest`. -These defaults can be overridden by setting the `startOffset` binding property. -There will be no committed offset(s) the first time the binding is started with a particular `group`. -The other condition where no committed offset exists is if the offset has been expired. -With modern brokers (since 2.1), and default broker properties, the offsets are expired 7 days after the last member leaves the group. -See the https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes[`offsets.retention.minutes`] broker property for more information. - -When `resetOffsets` is `true`, the binder applies similar semantics to those that apply when there is no committed offset on the broker, as if this binding has never consumed from the topic; i.e. any current committed offset is ignored. - -Following are two use cases when this might be used. - -1. Consuming from a compacted topic containing key/value pairs. -Set `resetOffsets` to `true` and `startOffset` to `earliest`; the binding will perform a `seekToBeginning` on all newly assigned partitions. - -2. Consuming from a topic containing events, where you are only interested in events that occur while this binding is running. -Set `resetOffsets` to `true` and `startOffset` to `latest`; the binding will perform a `seekToEnd` on all newly assigned partitions. - -IMPORTANT: If a rebalance occurs after the initial assignment, the seeks will only be performed on any newly assigned partitions that were not assigned during the initial assignment. - -For more control over topic offsets, see xref:kafka/kafka_overview/rebalance-listener.adoc[Using a KafkaBindingRebalanceListener]; when a listener is provided, `resetOffsets` should not be set to `true`, otherwise, that will cause an error. - -[[consuming-batches]] -== Consuming Batches - -Starting with version 3.0, when `spring.cloud.stream.bindings..consumer.batch-mode` is set to `true`, all of the records received by polling the Kafka `Consumer` will be presented as a `List` to the listener method. -Otherwise, the method will be called with one record at a time. -The size of the batch is controlled by Kafka consumer properties `max.poll.records`, `fetch.min.bytes`, `fetch.max.wait.ms`; refer to the Kafka documentation for more information. - -Starting with version `4.0.2`, the binder supports DLQ capabilities when consuming in batch mode. -Keep in mind that, when using DLQ on a consumer binding that is in batch mode, all the records received from the previous poll will be delivered to the DLQ topic. - -IMPORTANT: Retry within the binder is not supported when using batch mode, so `maxAttempts` will be overridden to 1. -You can configure a `DefaultErrorHandler` (using a `ListenerContainerCustomizer`) to achieve similar functionality to retry in the binder. -You can also use a manual `AckMode` and call `Ackowledgment.nack(index, sleep)` to commit the offsets for a partial batch and have the remaining records redelivered. -Refer to the https://docs.spring.io/spring-kafka/docs/2.3.15.BUILD-SNAPSHOT/reference/html/#committing-offsets[Spring for Apache Kafka documentation] for more information about these techniques. - [[kafka-producer-properties]] == Kafka Producer Properties @@ -437,7 +393,7 @@ The header contains a `RecordMetadata` object provided by the Kafka client; it i + `ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)` + -Failed sends go the producer error channel (if configured); see xref:kafka/kafka_overview/error-channels.adoc[Error Channels]. +Failed sends go the producer error channel (if configured); see <>. + Default: null. @@ -471,180 +427,3 @@ This property allows you to override that behavior. If set to true, records published to this output binding will not be run in a transaction, unless one is already in process. + Default: `false` - -[[usage-examples]] -== Usage examples - -In this section, we show the use of the preceding properties for specific scenarios. - -[[example:-setting-ackmode-to-manual-and-relying-on-manual-acknowledgement]] -=== Example: Setting `ackMode` to `MANUAL` and Relying on Manual Acknowledgement - -This example illustrates how one may manually acknowledge offsets in a consumer application. - -This example requires that `spring.cloud.stream.kafka.bindings.input.consumer.ackMode` be set to `MANUAL`. -Use the corresponding input channel name for your example. - -[source] ----- -@SpringBootApplication -public class ManuallyAcknowdledgingConsumer { - - public static void main(String[] args) { - SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); - } - - @Bean - public Consumer> process() { - return message -> { - Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); - if (acknowledgment != null) { - System.out.println("Acknowledgment provided"); - acknowledgment.acknowledge(); - } - }; -} ----- - -[[example:-security-configuration]] -=== Example: Security Configuration - -Apache Kafka 0.9 supports secure connections between client and brokers. -To take advantage of this feature, follow the guidelines in the https://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 https://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation]. -Use the `spring.cloud.stream.kafka.binder.configuration` option to set security properties for all clients created by the binder. - -For example, to set `security.protocol` to `SASL_SSL`, set the following property: - -[source] ----- -spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL ----- - -All the other security properties can be set in a similar manner. - -When using Kerberos, follow the instructions in the https://kafka.apache.org/090/documentation.html#security_sasl_clientconfig[reference documentation] for creating and referencing the JAAS configuration. - -Spring Cloud Stream supports passing JAAS configuration information to the application by using a JAAS configuration file and using Spring Boot properties. - -[[using-jaas-configuration-files]] -==== Using JAAS Configuration Files - -The JAAS and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties. -The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using a JAAS configuration file: - -[source,bash] ----- - java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \ - --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ - --spring.cloud.stream.bindings.input.destination=stream.ticktock \ - --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT ----- - -[[using-spring-boot-properties]] -==== Using Spring Boot Properties - -As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications by using Spring Boot properties. - -The following properties can be used to configure the login context of the Kafka client: - -spring.cloud.stream.kafka.binder.jaas.loginModule:: -The login module name. Not necessary to be set in normal cases. -+ -Default: `com.sun.security.auth.module.Krb5LoginModule`. -spring.cloud.stream.kafka.binder.jaas.controlFlag:: -The control flag of the login module. -+ -Default: `required`. -spring.cloud.stream.kafka.binder.jaas.options:: -Map with a key/value pair containing the login module options. -+ -Default: Empty map. - -The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using Spring Boot configuration properties: - -[source,bash] ----- - java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ - --spring.cloud.stream.bindings.input.destination=stream.ticktock \ - --spring.cloud.stream.kafka.binder.autoCreateTopics=false \ - --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \ - --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \ - --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \ - --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \ - --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM ----- - -The preceding example represents the equivalent of the following JAAS file: - -[source] ----- -KafkaClient { - com.sun.security.auth.module.Krb5LoginModule required - useKeyTab=true - storeKey=true - keyTab="/etc/security/keytabs/kafka_client.keytab" - principal="kafka-client-1@EXAMPLE.COM"; -}; ----- - -If the topics required already exist on the broker or will be created by an administrator, autocreation can be turned off and only client JAAS properties need to be sent. - -NOTE: Do not mix JAAS configuration files and Spring Boot properties in the same application. -If the `-Djava.security.auth.login.config` system property is already present, Spring Cloud Stream ignores the Spring Boot properties. - -NOTE: Be careful when using the `autoCreateTopics` and `autoAddPartitions` with Kerberos. -Usually, applications may use principals that do not have administrative rights in Kafka and Zookeeper. -Consequently, relying on Spring Cloud Stream to create/modify topics may fail. -In secure environments, we strongly recommend creating topics and managing ACLs administratively by using Kafka tooling. - -[[multi-binder-configuration-and-jaas]] -==== Multi-binder configuration and JAAS - -When connecting to multiple clusters in which each one requires separate JAAS configuration, then set the JAAS configuration using the property `sasl.jaas.config`. -When this property is present in the application, it takes precedence over the other strategies mentioned above. -See this https://cwiki.apache.org/confluence/display/KAFKA/KIP-85%3A+Dynamic+JAAS+configuration+for+Kafka+clients[KIP-85] for more details. - -For example, if you have two clusters in your application with separate JAAS configuration, then the following is a template that you can use: - -``` -spring.cloud.stream: - binders: - kafka1: - type: kafka - environment: - spring: - cloud: - stream: - kafka: - binder: - brokers: localhost:9092 - configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";" - kafka2: - type: kafka - environment: - spring: - cloud: - stream: - kafka: - binder: - brokers: localhost:9093 - configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user1\" password=\"user1-secret\";" - kafka.binder: - 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. - -See this https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/multi-binder-samples/kafka-multi-binder-jaas[sample application] for more details on how to setup and run such an application. - -[[pause-resume]] -=== Example: Pausing and Resuming the Consumer - -If you wish to suspend consumption but not cause a partition rebalance, you can pause and resume the consumer. -This is facilitated by managing the binding lifecycle as shown in **Binding visualization and control** in the Spring Cloud Stream documentation, using `State.PAUSED` and `State.RESUMED`. - -To resume, you can use an `ApplicationListener` (or `@EventListener` method) to receive `ListenerContainerIdleEvent` instances. -The frequency at which events are published is controlled by the `idleEventInterval` property. - diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/consumer-producer-config-customizer.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/cons-prod-config-cust.adoc similarity index 96% rename from docs/modules/ROOT/pages/kafka/kafka_overview/consumer-producer-config-customizer.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/cons-prod-config-cust.adoc index 2e48486cc..aa91af3ea 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/consumer-producer-config-customizer.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/cons-prod-config-cust.adoc @@ -1,6 +1,5 @@ [[consumer-producer-config-customizer]] = Customizing Consumer and Producer configuration -:page-section-summary-toc: 1 If you want advanced customization of consumer and producer configuration that is used for creating `ConsumerFactory` and `ProducerFactory` in Kafka, you can implement the following customizers. @@ -13,4 +12,3 @@ For example, if you want to gain access to a bean that is defined at the applica When the binder discovers that these customizers are available as beans, it will invoke the `configure` method right before creating the consumer and producer factories. Both of these interfaces also provide access to both the binding and destination names so that they can be accessed while customizing producer and consumer properties. - diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc new file mode 100644 index 000000000..9ad5856bb --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/consume-batches.adoc @@ -0,0 +1,13 @@ += Consuming Batches + +Starting with version 3.0, when `spring.cloud.stream.bindings..consumer.batch-mode` is set to `true`, all of the records received by polling the Kafka `Consumer` will be presented as a `List` to the listener method. +Otherwise, the method will be called with one record at a time. +The size of the batch is controlled by Kafka consumer properties `max.poll.records`, `fetch.min.bytes`, `fetch.max.wait.ms`; refer to the Kafka documentation for more information. + +Starting with version `4.0.2`, the binder supports DLQ capabilities when consuming in batch mode. +Keep in mind that, when using DLQ on a consumer binding that is in batch mode, all the records received from the previous poll will be delivered to the DLQ topic. + +IMPORTANT: Retry within the binder is not supported when using batch mode, so `maxAttempts` will be overridden to 1. +You can configure a `DefaultErrorHandler` (using a `ListenerContainerCustomizer`) to achieve similar functionality to retry in the binder. +You can also use a manual `AckMode` and call `Ackowledgment.nack(index, sleep)` to commit the offsets for a partial batch and have the remaining records redelivered. +Refer to the https://docs.spring.io/spring-kafka/docs/2.3.15.BUILD-SNAPSHOT/reference/html/#committing-offsets[Spring for Apache Kafka documentation] for more information about these techniques. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc similarity index 58% rename from docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc index 3aa1f4bb3..f956901cd 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator-example.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/custom-health-ind.adoc @@ -1,5 +1,31 @@ +[[custom-kafka-binder-health-indicator]] += Custom Kafka Binder Health Indicator + +== Overriding Default Kafka Binder Health Indicator + +Kafka binder activates a default health indicator when Spring Boot actuator is on the classpath. +This health indicator checks the health of the binder and any communication issues with the Kafka broker. +If an application wants to disable this default health check implementation and include a custom implementation, then it can provide an implementation for `KafkaBinderHealth` interface. +`KafkaBinderHealth` is a marker interface that extends from `HealthIndicator`. +In the custom implementation, it must provide an implementation for the `health()` method. +The custom implementation must be present in the application configuration as a bean. +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. + +``` +@Bean +public KafkaBinderHealth kafkaBinderHealthIndicator() { + return new KafkaBinderHealth() { + @Override + public Health health() { + // custom implementation details. + } + }; +} +``` + [[custom-kafka-binder-health-indicator-example]] -= Custom kafka Binder Health Indicator Example +== Custom kafka Binder Health Indicator Example 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. @@ -42,7 +68,6 @@ public class KafkaBinderHealthImplementation implements KafkaBinderHealth { } ``` - Then we need to create a bean for the custom implementation. ``` diff --git a/docs/modules/ROOT/pages/kafka/kafka_dlq.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/dlq.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/kafka_dlq.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/dlq.adoc diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/error-channels.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/error-channels.adoc similarity index 90% rename from docs/modules/ROOT/pages/kafka/kafka_overview/error-channels.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/error-channels.adoc index cfc6faa9d..d2eec0490 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/error-channels.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/error-channels.adoc @@ -1,6 +1,5 @@ [[kafka-error-channels]] = Error Channels -:page-section-summary-toc: 1 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 https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-error-handling[this section on error handling] for more information. @@ -10,6 +9,5 @@ The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureExcep * `failedMessage`: The Spring Messaging `Message` that failed to be sent. * `record`: The raw `ProducerRecord` that was created from the `failedMessage` -There is no automatic handling of producer exceptions (such as sending to a xref:kafka/kafka_dlq.adoc[Dead-Letter queue]). +There is no automatic handling of producer exceptions (such as sending to a <>). You can consume these exceptions with your own Spring Integration flow. - diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc new file mode 100644 index 000000000..c200cae66 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc @@ -0,0 +1,27 @@ += Manual Acknowledgement + +This example illustrates how one may manually acknowledge offsets in a consumer application. + +This example requires that `spring.cloud.stream.kafka.bindings.input.consumer.ackMode` be set to `MANUAL`. +Use the corresponding input channel name for your example. + +[source] +---- +@SpringBootApplication +public class ManuallyAcknowdledgingConsumer { + + public static void main(String[] args) { + SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); + } + + @Bean + public Consumer> process() { + return message -> { + Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); + if (acknowledgment != null) { + System.out.println("Acknowledgment provided"); + acknowledgment.acknowledge(); + } + }; +} +---- diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/metrics.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc similarity index 90% rename from docs/modules/ROOT/pages/kafka/kafka_overview/metrics.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc index 56d0e4528..cf08ae244 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/metrics.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/metrics.adoc @@ -10,7 +10,7 @@ The metric contains the consumer group information, topic and the actual lag in This metric is particularly useful for providing auto-scaling feedback to a PaaS platform. The metric collection behaviour can be configured by setting properties in the `spring.cloud.stream.kafka.binder.metrics` namespace, -refer to the xref:kafka/kafka_overview/configuration-options.adoc#kafka-binder-properties[kafka binder properties section] for more information. +refer to the <> for more information. You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application. @@ -25,4 +25,3 @@ class NoOpBindingMeters { ``` 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_overview/overview.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/overview.adoc similarity index 94% rename from docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/overview.adoc index a985f89e9..a341998b4 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/overview.adoc @@ -1,6 +1,4 @@ -[[overview]] = Overview -:page-section-summary-toc: 1 The following image shows a simplified diagram of how the Apache Kafka binder operates: @@ -15,4 +13,3 @@ The binder currently uses the Apache Kafka `kafka-clients` version `3.1.0`. This client can communicate with older brokers (see the Kafka documentation), but certain features may not be available. For example, with versions earlier than 0.11.x.x, native headers are not supported. Also, 0.11.x.x does not support the `autoAddPartitions` property. - diff --git a/docs/modules/ROOT/pages/kafka/kafka_partitions.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/partitions.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/kafka_partitions.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/partitions.adoc diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/pause_resume.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/pause_resume.adoc new file mode 100644 index 000000000..b502e5023 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/pause_resume.adoc @@ -0,0 +1,8 @@ +[[pause-resume]] += Pausing and Resuming the Consumer + +If you wish to suspend consumption but not cause a partition rebalance, you can pause and resume the consumer. +This is facilitated by managing the binding lifecycle as shown in **Binding visualization and control** in the Spring Cloud Stream documentation, using `State.PAUSED` and `State.RESUMED`. + +To resume, you can use an `ApplicationListener` (or `@EventListener` method) to receive `ListenerContainerIdleEvent` instances. +The frequency at which events are published is controlled by the `idleEventInterval` property. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/rebalance-listener.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/rebalance_listener.adoc similarity index 97% rename from docs/modules/ROOT/pages/kafka/kafka_overview/rebalance-listener.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/rebalance_listener.adoc index fa372ad77..247608f8d 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/rebalance-listener.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/rebalance_listener.adoc @@ -1,9 +1,10 @@ [[rebalance-listener]] -= Using a KafkaBindingRebalanceListener += KafkaBindingRebalanceListener Applications may wish to seek topics/partitions to arbitrary offsets when the partitions are initially assigned, or perform other operations on the consumer. Starting with version 2.1, if you provide a single `KafkaBindingRebalanceListener` bean in the application context, it will be wired into all Kafka consumer bindings. +==== [source, java] ---- public interface KafkaBindingRebalanceListener { @@ -44,6 +45,6 @@ public interface KafkaBindingRebalanceListener { } ---- +==== You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener. - diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/reset-offsets.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/reset-offsets.adoc new file mode 100644 index 000000000..5cd99c5c2 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/reset-offsets.adoc @@ -0,0 +1,26 @@ +[[reset-offsets]] += Resetting Offsets + +When an application starts, the initial position in each assigned partition depends on two properties `startOffset` and `resetOffsets`. +If `resetOffsets` is `false`, normal Kafka consumer https://kafka.apache.org/documentation/#consumerconfigs_auto.offset.reset[`auto.offset.reset`] semantics apply. +i.e. If there is no committed offset for a partition for the binding's consumer group, the position is `earliest` or `latest`. +By default, bindings with an explicit `group` use `earliest`, and anonymous bindings (with no `group`) use `latest`. +These defaults can be overridden by setting the `startOffset` binding property. +There will be no committed offset(s) the first time the binding is started with a particular `group`. +The other condition where no committed offset exists is if the offset has been expired. +With modern brokers (since 2.1), and default broker properties, the offsets are expired 7 days after the last member leaves the group. +See the https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes[`offsets.retention.minutes`] broker property for more information. + +When `resetOffsets` is `true`, the binder applies similar semantics to those that apply when there is no committed offset on the broker, as if this binding has never consumed from the topic; i.e. any current committed offset is ignored. + +Following are two use cases when this might be used. + +1. Consuming from a compacted topic containing key/value pairs. +Set `resetOffsets` to `true` and `startOffset` to `earliest`; the binding will perform a `seekToBeginning` on all newly assigned partitions. + +2. Consuming from a topic containing events, where you are only interested in events that occur while this binding is running. +Set `resetOffsets` to `true` and `startOffset` to `latest`; the binding will perform a `seekToEnd` on all newly assigned partitions. + +IMPORTANT: If a rebalance occurs after the initial assignment, the seeks will only be performed on any newly assigned partitions that were not assigned during the initial assignment. + +For more control over topic offsets, see <>; when a listener is provided, `resetOffsets` should not be set to `true`, otherwise, that will cause an error. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/retry-and-dlq-processing.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/retry-dlq.adoc similarity index 97% rename from docs/modules/ROOT/pages/kafka/kafka_overview/retry-and-dlq-processing.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/retry-dlq.adoc index 63c61c396..4d814c90d 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/retry-and-dlq-processing.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/retry-dlq.adoc @@ -13,6 +13,7 @@ There are situations where it is preferable to move this functionality to the li To configure moving this functionality from the binder to the container, define a `@Bean` of type `ListenerContainerWithDlqAndRetryCustomizer`. This interface has the following methods: +==== [source, java] ---- /** @@ -42,9 +43,11 @@ default boolean retryAndDlqInBinding(String destinationName, String group) { return true; } ---- +==== The destination resolver and `BackOff` are created from the binding properties (if configured). The `KafkaTemplate` uses configuration from `spring.kafka....` properties. You can then use these to create a custom error handler and dead letter publisher; for example: +==== [source, java] ---- @Bean @@ -72,11 +75,13 @@ ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate template) { }; } ---- +==== Now, only a single retry delay needs to be greater than the consumer's `max.poll.interval.ms` property. When working with several binders, the 'ListenerContainerWithDlqAndRetryCustomizer' bean gets overridden by the 'DefaultBinderFactory'. For the bean -to apply, you need to use a 'BinderCustomizer' to set the container customizer (See xref:spring-cloud-stream/binder-customizer.adoc[Customizing binders in multi binder applications]): +to apply, you need to use a 'BinderCustomizer' to set the container customizer (See <>): +==== [source, java] ---- @Bean @@ -94,4 +99,4 @@ public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomi }; } ---- - +==== diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc new file mode 100644 index 000000000..6dbe6b5d0 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/security-config.adoc @@ -0,0 +1,128 @@ += Security Configuration + +Apache Kafka supports secure connections between client and brokers. +To take advantage of this feature, follow the guidelines in the https://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 https://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation]. +Use the `spring.cloud.stream.kafka.binder.configuration` option to set security properties for all clients created by the binder. + +For example, to set `security.protocol` to `SASL_SSL`, set the following property: + +[source] +---- +spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL +---- + +All the other security properties can be set in a similar manner. + +When using Kerberos, follow the instructions in the https://kafka.apache.org/090/documentation.html#security_sasl_clientconfig[reference documentation] for creating and referencing the JAAS configuration. + +Spring Cloud Stream supports passing JAAS configuration information to the application by using a JAAS configuration file and using Spring Boot properties. + +== Using JAAS Configuration Files + +The JAAS and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties. +The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using a JAAS configuration file: + +[source,bash] +---- + java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \ + --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ + --spring.cloud.stream.bindings.input.destination=stream.ticktock \ + --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT +---- + +== Using Spring Boot Properties + +As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications by using Spring Boot properties. + +The following properties can be used to configure the login context of the Kafka client: + +spring.cloud.stream.kafka.binder.jaas.loginModule:: +The login module name. Not necessary to be set in normal cases. ++ +Default: `com.sun.security.auth.module.Krb5LoginModule`. +spring.cloud.stream.kafka.binder.jaas.controlFlag:: +The control flag of the login module. ++ +Default: `required`. +spring.cloud.stream.kafka.binder.jaas.options:: +Map with a key/value pair containing the login module options. ++ +Default: Empty map. + +The following example shows how to launch a Spring Cloud Stream application with SASL and Kerberos by using Spring Boot configuration properties: + +[source,bash] +---- + java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ + --spring.cloud.stream.bindings.input.destination=stream.ticktock \ + --spring.cloud.stream.kafka.binder.autoCreateTopics=false \ + --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \ + --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \ + --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \ + --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \ + --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM +---- + +The preceding example represents the equivalent of the following JAAS file: + +[source] +---- +KafkaClient { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab="/etc/security/keytabs/kafka_client.keytab" + principal="kafka-client-1@EXAMPLE.COM"; +}; +---- + +If the topics required already exist on the broker or will be created by an administrator, autocreation can be turned off and only client JAAS properties need to be sent. + +NOTE: Do not mix JAAS configuration files and Spring Boot properties in the same application. +If the `-Djava.security.auth.login.config` system property is already present, Spring Cloud Stream ignores the Spring Boot properties. + +NOTE: Be careful when using the `autoCreateTopics` and `autoAddPartitions` with Kerberos. +Usually, applications may use principals that do not have administrative rights in Kafka and Zookeeper. +Consequently, relying on Spring Cloud Stream to create/modify topics may fail. +In secure environments, we strongly recommend creating topics and managing ACLs administratively by using Kafka tooling. + +== Multi-binder configuration and JAAS + +When connecting to multiple clusters in which each one requires separate JAAS configuration, then set the JAAS configuration using the property `sasl.jaas.config`. +When this property is present in the application, it takes precedence over the other strategies mentioned above. +See this https://cwiki.apache.org/confluence/display/KAFKA/KIP-85%3A+Dynamic+JAAS+configuration+for+Kafka+clients[KIP-85] for more details. + +For example, if you have two clusters in your application with separate JAAS configuration, then the following is a template that you can use: + +``` +spring.cloud.stream: + binders: + kafka1: + type: kafka + environment: + spring: + cloud: + stream: + kafka: + binder: + brokers: localhost:9092 + configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";" + kafka2: + type: kafka + environment: + spring: + cloud: + stream: + kafka: + binder: + brokers: localhost:9093 + configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user1\" password=\"user1-secret\";" + kafka.binder: + 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. + +See this https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/multi-binder-samples/kafka-multi-binder-jaas[sample application] for more details on how to setup and run such an application. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/tombstones.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/tombstone.adoc similarity index 90% rename from docs/modules/ROOT/pages/kafka/kafka_overview/tombstones.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/tombstone.adoc index b9859c9ad..709fd5c09 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/tombstones.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/tombstone.adoc @@ -1,10 +1,10 @@ [[kafka-tombstones]] -= Tombstone Records (null record values) -:page-section-summary-toc: 1 += Tombstone Records When using compacted topics, a record with a `null` value (also called a tombstone record) represents the deletion of a key. To receive such messages in a Spring Cloud Stream function, you can use the following strategy. +==== [source, java] ---- @Bean @@ -21,4 +21,4 @@ public Function, String> myFunction() { }; } ---- - +==== diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/transactional-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/transactional.adoc similarity index 99% rename from docs/modules/ROOT/pages/kafka/kafka_overview/transactional-binder.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/transactional.adoc index e549ae454..c1d5a09eb 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/transactional-binder.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/transactional.adoc @@ -12,6 +12,7 @@ Similarly, instead of publishing dead-letter records within the transaction, thi If you wish to use transactions in a source application, or from some arbitrary thread for producer-only transaction (e.g. `@Scheduled` method), you must get a reference to the transactional producer factory and define a `KafkaTransactionManager` bean using it. +==== [source, java] ---- @Bean @@ -25,6 +26,7 @@ public PlatformTransactionManager transactionManager(BinderFactory binders, return tm; } ---- +==== Notice that we get a reference to the binder using the `BinderFactory`; use `null` in the first argument when there is only one binder configured. If more than one binder is configured, use the binder name to get the reference. @@ -32,6 +34,7 @@ Once we have a reference to the binder, we can obtain a reference to the `Produc Then you would use normal Spring transaction support, e.g. `TransactionTemplate` or `@Transactional`, for example: +==== [source, java] ---- public static class Sender { @@ -43,8 +46,8 @@ public static class Sender { } ---- +==== If you wish to synchronize producer-only transactions with those from some other transaction manager, use a `ChainedTransactionManager`. IMPORTANT: If you deploy multiple instances of your application, each instance needs a unique `transactionIdPrefix`. - diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/usage.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/usage.adoc similarity index 93% rename from docs/modules/ROOT/pages/kafka/kafka_overview/usage.adoc rename to docs/modules/ROOT/pages/kafka/kafka-binder/usage.adoc index 756799ac7..0b58c4368 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/usage.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/usage.adoc @@ -1,6 +1,4 @@ -[[usage]] = Usage -:page-section-summary-toc: 1 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: @@ -21,4 +19,3 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown spring-cloud-starter-stream-kafka ---- - diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc new file mode 100644 index 000000000..46f74aef4 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/concurrency.adoc @@ -0,0 +1,15 @@ +[[concurrency]] += Concurrency + +When using reactive functions with the reactive Kafka binder, if you set concurrency on the consumer binding, then the binder creates as many dedicated `KafkaReceiver` objects as provided by the concurrency value. +In other words, this creates multiple reactive streams with separate `Flux` implementations. +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. + +``` +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 new file mode 100644 index 000000000..bd521b30d --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/consuming.adoc @@ -0,0 +1,62 @@ +[[consuming-records-in-the-raw-format]] += Consuming Records + +In the above `upppercase` function, we are consuming the record as `Flux` and then produce it as `Flux`. +There might be occasions in which you need to receive the record in the original received format - the `ReceiverRecord`. +Here is such a function. + +[source, java] +---- +@Bean +public Function>, Flux> lowercase() { + return s -> s.map(rec -> new String(rec.value()).toLowerCase()); +} +---- + +In this function, note that, we are consuming the record as `Flux>` and then producing it as `Flux`. +`ReceiverRecord` is the basic received record which is a specialized Kafka `ConsumerRecord` in Reactor Kafka. +When using the reactive Kafka binder, the above function will give you access to the `ReceiverRecord` type for each incoming record. +However, in this case, you need to provide a custom implementation for a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/RecordMessageConverter.html[RecordMessageConverter]. +By default, the reactive Kafka binder uses a https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/support/converter/MessagingMessageConverter.html[MessagingMessageConverter] that converts the payload and headers from the `ConsumerRecord`. +Therefore, by the time your handler method receives it, the payload is already extracted from the received record and passed onto the method as in the case of the first function we looked above. +By providing a custom `RecordMessageConverter` implementation in the application, you can override the default behavior. +For example, if you want to consume the record as raw `Flux>`, then you can provide the following bean definition in the application. + +[source, java] +---- +@Bean +RecordMessageConverter fullRawReceivedRecord() { + return new RecordMessageConverter() { + + private final RecordMessageConverter converter = new MessagingMessageConverter(); + + @Override + public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment, + Consumer consumer, Type payloadType) { + return MessageBuilder.withPayload(record).build(); + } + + @Override + public ProducerRecord fromMessage(Message message, String defaultTopic) { + return this.converter.fromMessage(message, defaultTopic); + } + + }; +} +---- + +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" +``` + +`lowercase-in-0` is the input binding name for our `lowercase` function. +For the outbound (`lowecase-out-0`), we still use the regular `MessagingMessageConverter`. + +In the `toMessage` implementation above, we receive the raw `ConsumerRecord` (`ReceiverRecord` since we are in a reactive binder context) and then wrap it inside a `Message`. +Then that message payload which is the `ReceiverRecord` is provided to the user method. + +If `reactiveAutoCommit` is `false` (default), call `rec.receiverOffset().acknowledge()` (or `commit()`) to cause the offset to be committed; if `reactiveAutoCommit` is `true`, the flux supplies `ConsumerRecord` s instead. +Refer to the `reactor-kafka` documentation and javadocs for more information. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/examples.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/examples.adoc new file mode 100644 index 000000000..ecd10286f --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/examples.adoc @@ -0,0 +1,71 @@ +[[basic-example-using-the-reactive-kafka-binder]] += Basic Example using the Reactive Kafka Binder + +In this section, we show some basic code snippets for writing a reactive Kafka application using the reactive binder and details around them. + +[source, java] +---- +@Bean +public Function, Flux> uppercase() { + return s -> s.map(String::toUpperCase); +} +---- + +You can use the above `upppercase` function with both message channel based Kafka binder (`spring-cloud-stream-binder-kafka`) as well as the reactive Kafka binder (`spring-cloud-stream-binder-kafka-reactive`), the topic of discussion in this section. +When using this function with the regular Kafka binder, although you are using reactive types in the application (i.e., in the `uppercase` function), you only get the reactive streams within the execution of your function. +Outside the function's execution context, there is no reactive benefits since the underlying binder is not based on the reactive stack. +Therefore, although this might look like it is bringing a full end-to-end reactive stack, this application is only partially reactive. + +Now assume that you are using the proper reactive binder for Kafka - `spring-cloud-stream-binder-kafka-reactive` with the above function's application. +This binder implementation will give the full reactive benefits all the way from consumption on the top end to publishing at the bottom end of the chain. +This is because the underlying binder is built on top of https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka]'s core API's. +On the consumer side, it makes use of the https://projectreactor.io/docs/kafka/release/reference/#api-guide-receiver[KafkaReceiver] which is a reactive implementation of a Kafka consumer. +Similarly, on the producer side, it uses https://projectreactor.io/docs/kafka/release/reference/#api-guide-sender[KafkaSender] API which is the reactive implementation of a Kafka producer. +Since the foundations of the reactive Kafka binder is built upon a proper reactive Kafka API, applications get the full benefits of using reactive technologies. +Things like automatic back pressure, among other reactive capabilities, are built-in for the application when using this reactive Kafka binder. + +Starting with version 4.0.2, you can customize the `ReceiverOptions` and `SenderOptions` by providing one or more `ReceiverOptionsCustomizer` or `SenderOptionsCustomizer` beans respectively. +They are `BiFunction` s which receive the binding name and initial options, returning the customized options. +The interfaces extend `Ordered` so the customizers will be applied in the order required, when more than one are present. + +IMPORTANT: The binder does not commit offsets by default. +Starting with version 4.0.2, the `KafkaHeaders.ACKNOWLEDGMENT` header contains a `ReceiverOffset` object which allows you to cause the offset to be committed by calling its `acknowledge()` or `commit()` methods. + +[source, java] +---- +@Bean +public Consumer> consume() { + return msg -> { + process(msg.getPayload()); + msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge(); + } +} +---- + +Refer to the `reactor-kafka` documentation and javadocs for more information. + +In addition, starting with version 4.0.3, the Kafka consumer property `reactiveAtmostOnce` can be set to `true` and the binder will automatically commit the offsets before records returned by each poll are processed. +Also, starting with version 4.0.3, you can set the consumer property `reactiveAutoCommit` to `true` and the the binder will automatically commit the offsets after the records returned by each poll are processed. +In these cases, the acknowledgment header is not present. + +IMPORTANT: 4.0.2 also provided `reactiveAutoCommit`, but the implementation was incorrect, it behaved similarly to `reactiveAtMostOnce`. + +The following is an example of how to use `reaciveAutoCommit`. + +[source, java] +---- +@Bean +Consumer>>> input() { + return flux -> flux + .doOnNext(inner -> inner + .doOnNext(val -> { + log.info(val.value()); + }) + .subscribe()) + .subscribe(); +} +---- + +Note that `reactor-kafka` returns a `Flux>>` when using auto commit. +Given that Spring has no access to the contents of the inner flux, the application must deal with the native `ConsumerRecord`; there is no message conversion or conversion service applied to the contents. +This requires the use of native decoding (by specifying a `Deserializer` of the appropriate type in the configuration) to return record keys/values of the desired types. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/multiplex.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/multiplex.adoc new file mode 100644 index 000000000..c7ce987a9 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/multiplex.adoc @@ -0,0 +1,5 @@ +[[multiplex]] += Multiplex + +Starting with version 4.0.3, the common consumer property `multiplex` is now supported by the reactive binder, where a single binding can consume from multiple topics. +When `false` (default), a separate binding is created for each topic specified in a comma-delimited list in the common `destination` property. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/overview.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/overview.adoc new file mode 100644 index 000000000..48c6a5d8b --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/overview.adoc @@ -0,0 +1,6 @@ +[[reactive-kafka-binder]] += Overview + +Kafka binder ecosystem in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project. +This reactive Kafka binder enables full end-to-end reactive capabilities such as backpressure, reactive streams etc. in applications based on Apache Kafka. +When your Spring Cloud Stream Kafka application is written using reactive types (`Flux`, `Mono` etc.), it is recommended to use this reactive Kafka binder instead of the regular message channel based Kafka binder. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/pattern.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/pattern.adoc new file mode 100644 index 000000000..540361083 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/pattern.adoc @@ -0,0 +1,5 @@ +[[destination-is-pattern]] += Destination is Pattern + +Starting with version 4.0.3, the `destination-is-pattern` Kafka binding consumer property is now supported. +The receiver options are conigured with a regex `Pattern`, allowing the binding to consume from any topic that matches the pattern. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/sender_result.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/sender_result.adoc new file mode 100644 index 000000000..225d0e137 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/sender_result.adoc @@ -0,0 +1,41 @@ +[[sender-result-channel]] += Sender Result Channel + +Starting with version 4.0.3, you can configure the `resultMetadataChannel` to receive `SenderResult` s to determine success/failure of sends. + +The `SenderResult` contains `correlationMetadata` to allow you to correlate results with sends; it also contains `RecordMetadata`, which indicates the `TopicPartition` and offset of the sent record. + +The `resultMetadataChannel` **must** be a `FluxMessageChannel` instance. + +Here is an example of how to use this feature, with correlation metadata of type `Integer`: + +[source, java] +---- +@Bean +FluxMessageChannel sendResults() { + return new FluxMessageChannel(); +} + +@ServiceActivator(inputChannel = "sendResults") +void handleResults(SenderResult result) { + if (result.exception() != null) { + failureFor(result); + } + else { + successFor(result); + } +} +---- + +To set the correlation metadata on an output record, set the `CORRELATION_ID` header: + +[source, java] +---- +streamBridge.send("words1", MessageBuilder.withPayload("foobar") + .setCorrelationId(42) + .build()); +---- + +When using the feature with a `Function`, the function output type must be a `Message` with the correlation id header set to the desired value. + +Metadata should be unique, at least for the duration of the send. diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc new file mode 100644 index 000000000..fed5e0f83 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/usage.adoc @@ -0,0 +1,11 @@ +[[maven-coordinates]] += Maven Coordinates + +Following are the maven coordinates for the reactive Kafka binder. + +``` + + org.springframework.cloud + spring-cloud-stream-binder-kafka-reactive + +``` diff --git a/docs/modules/ROOT/pages/kafka/streams/accessing-metrics.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/accessing-metrics.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/accessing-metrics.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/accessing-metrics.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/ancillaries-to-the-programming-model.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/ancillaries-to-the-programming-model.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/ancillaries-to-the-programming-model.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/binding-visualization-and-control-in-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/binding-visualization-and-control-in-binder.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/binding-visualization-and-control-in-binder.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/configuration-options.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/configuration-options.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/error-handling.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/error-handling.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/error-handling.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/event-type-based-routing-in-applications.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/event-type-based-routing-in-applications.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/event-type-based-routing-in-applications.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/event-type-based-routing-in-applications.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/health-indicator.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/health-indicator.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/health-indicator.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/health-indicator.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/interactive-queries.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/interactive-queries.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/interactive-queries.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/manually-starting-processors-selectively.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/manually-starting-processors-selectively.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors-selectively.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/manually-starting-processors.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/manually-starting-processors.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/manually-starting-processors.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/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 similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/mixing-high-level-dsl-and-low-level-processor-api.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/mixing-high-level-dsl-and-low-level-processor-api.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/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 similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/multi-binders-with-based-binders-and-regular-binder.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/overview.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/overview.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/overview.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/overview.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/partition-support-on-the-outbound.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/partition-support-on-the-outbound.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/partition-support-on-the-outbound.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/programming-model.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/programming-model.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/record-serialization-and-deserialization.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/record-serialization-and-deserialization.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/record-serialization-and-deserialization.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/record-serialization-and-deserialization.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/retrying-critical-business-logic.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/retrying-critical-business-logic.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/retrying-critical-business-logic.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/state-cleanup.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/state-cleanup.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/state-cleanup.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/state-cleanup.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/state-store.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/state-store.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/state-store.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/state-store.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/streamsbuilderfactorybean-customizer.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/streamsbuilderfactorybean-customizer.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/streamsbuilderfactorybean-customizer.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/timestamp-extractor.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/timestamp-extractor.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/timestamp-extractor.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/topology-visualization.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/topology-visualization.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/topology-visualization.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/topology-visualization.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/tracing-using-spring-cloud-sleuth.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/tracing-using-spring-cloud-sleuth.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/tracing-using-spring-cloud-sleuth.adoc diff --git a/docs/modules/ROOT/pages/kafka/streams/usage.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/usage.adoc similarity index 100% rename from docs/modules/ROOT/pages/kafka/streams/usage.adoc rename to docs/modules/ROOT/pages/kafka/kafka-streams-binder/usage.adoc diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index 53e0a8cd0..4b8d56246 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc @@ -1,7 +1,3 @@ -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 === Usage @@ -806,7 +802,7 @@ public Function, String> myFunction() { ==== [[rebalance-listener]] -=== Using a KafkaBindingRebalanceListener +=== KafkaBindingRebalanceListener Applications may wish to seek topics/partitions to arbitrary offsets when the partitions are initially assigned, or perform other operations on the consumer. Starting with version 2.1, if you provide a single `KafkaBindingRebalanceListener` bean in the application context, it will be wired into all Kafka consumer bindings. diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator.adoc deleted file mode 100644 index f8f8506d5..000000000 --- a/docs/modules/ROOT/pages/kafka/kafka_overview/custom-binder-health-indicator.adoc +++ /dev/null @@ -1,25 +0,0 @@ -[[custom-kafka-binder-health-indicator]] -= Custom Kafka Binder Health Indicator -:page-section-summary-toc: 1 - -Kafka binder activates a default health indicator when Spring Boot actuator is on the classpath. -This health indicator checks the health of the binder and any communication issues with the Kafka broker. -If an application wants to disable this default health check implementation and include a custom implementation, then it can provide an implementation for `KafkaBinderHealth` interface. -`KafkaBinderHealth` is a marker interface that extends from `HealthIndicator`. -In the custom implementation, it must provide an implementation for the `health()` method. -The custom implementation must be present in the application configuration as a bean. -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. - -``` -@Bean -public KafkaBinderHealth kafkaBinderHealthIndicator() { - return new KafkaBinderHealth() { - @Override - public Health health() { - // custom implementation details. - } - }; -} -``` - diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive.adoc b/docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc similarity index 98% rename from docs/modules/ROOT/pages/kafka/kafka-reactive.adoc rename to docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc index 1460fdc8b..c3b0a879e 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-reactive.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_reactive_binder.adoc @@ -1,7 +1,7 @@ [[reactive-kafka-binder]] = Reactive Kafka Binder -Kafka binder in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project. +Kafka binder ecosystem in Spring Cloud Stream provides a dedicated reactive binder based on the https://projectreactor.io/docs/kafka/release/reference/[Reactor Kafka] project. This reactive Kafka binder enables full end-to-end reactive capabilities such as backpressure, reactive streams etc. in applications based on Apache Kafka. When your Spring Cloud Stream Kafka application is written using reactive types (`Flux`, `Mono` etc.), it is recommended to use this reactive Kafka binder instead of the regular message channel based Kafka binder.