From 81f4a861c58f843274bf4842fc72cedb651460ba Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 19 Nov 2018 16:29:16 +0100 Subject: [PATCH] Fixed Kafka Stream binder docs --- README.adoc | 84 +++++++++++++++---- docs/src/main/asciidoc/dlq.adoc | 2 +- docs/src/main/asciidoc/kafka-streams.adoc | 39 +++++---- docs/src/main/asciidoc/overview.adoc | 34 ++++---- docs/src/main/asciidoc/partitions.adoc | 2 +- .../spring-cloud-stream-binder-kafka.adoc | 2 + 6 files changed, 112 insertions(+), 51 deletions(-) diff --git a/README.adoc b/README.adoc index 155660523..a12cdf7ad 100644 --- a/README.adoc +++ b/README.adoc @@ -21,7 +21,9 @@ It contains information about its design, usage, and configuration options, as w In addition, this guide explains the Kafka Streams binding capabilities of Spring Cloud Stream. -- -== Usage +== Apache Kafka Binder + +=== Usage To use Apache Kafka binder, you need to add `spring-cloud-stream-binder-kafka` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven: @@ -43,7 +45,7 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown ---- -== Apache Kafka Binder Overview +=== Overview The following image shows a simplified diagram of how the Apache Kafka binder operates: @@ -59,13 +61,13 @@ This client can communicate with older brokers (see the Kafka documentation), bu 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. -== Configuration Options +=== Configuration Options This section contains the configuration options used by the Apache Kafka binder. For common configuration options and properties pertaining to binder, see the <>. -=== Kafka Binder Properties +==== Kafka Binder Properties spring.cloud.stream.kafka.binder.brokers:: A list of brokers to which the Kafka binder connects. @@ -154,7 +156,7 @@ Use this, for example, if you wish to customize the trusted packages in a `Defau Default: none. [[kafka-consumer-properties]] -=== Kafka Consumer Properties +==== Kafka Consumer Properties The following properties are available for Kafka consumers only and must be prefixed with `spring.cloud.stream.kafka.bindings..consumer.`. @@ -210,6 +212,7 @@ If not set (the default), it effectively has the same value as `enableDlq`, auto Default: not set. resetOffsets:: Whether to reset offsets on the consumer to the value provided by startOffset. +Must be false if a `KafkaRebalanceListener` is provided; see <>. + Default: `false`. startOffset:: @@ -269,7 +272,7 @@ This can be configured using the `configuration` property above. Default: `false` [[kafka-producer-properties]] -=== Kafka Producer Properties +==== Kafka Producer Properties The following properties are available for Kafka producers only and must be prefixed with `spring.cloud.stream.kafka.bindings..producer.`. @@ -330,11 +333,11 @@ If a topic already exists with a smaller partition count and `autoAddPartitions` If a topic already exists with a smaller partition count and `autoAddPartitions` is enabled, new partitions are added. If a topic already exists with a larger number of partitions than the maximum of (`minPartitionCount` or `partitionCount`), the existing partition count is used. -=== Usage examples +==== Usage examples In this section, we show the use of the preceding properties for specific scenarios. -==== Example: Setting `autoCommitOffset` to `false` and Relying on Manual Acking +===== Example: Setting `autoCommitOffset` to `false` and Relying on Manual Acking This example illustrates how one may manually acknowledge offsets in a consumer application. @@ -362,7 +365,7 @@ public class ManuallyAcknowdledgingConsumer { } ---- -==== 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 http://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 http://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation]. @@ -381,7 +384,7 @@ When using Kerberos, follow the instructions in the http://kafka.apache.org/090/ 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: @@ -394,7 +397,7 @@ The following example shows how to launch a Spring Cloud Stream application with --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. @@ -451,7 +454,7 @@ 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. [[pause-resume]] -==== Example: Pausing and Resuming the Consumer +===== 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 adding the `Consumer` as a parameter to your `@StreamListener`. @@ -491,7 +494,7 @@ public class Application { ---- [[kafka-error-channels]] -== Error Channels +=== Error Channels Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel. See <> for more information. @@ -505,7 +508,7 @@ There is no automatic handling of producer exceptions (such as sending to a < consumer, + Collection partitions) { + + } + + /** + * Invoked by the container after any pending offsets are committed. + * @param bindingName the name of the binding. + * @param consumer the consumer. + * @param partitions the partitions. + */ + default void onPartitionsRevokedAfterCommit(String bindingName, Consumer consumer, Collection partitions) { + + } + + /** + * Invoked when partitions are initially assigned or after a rebalance. + * Applications might only want to perform seek operations on an initial assignment. + * @param bindingName the name of the binding. + * @param consumer the consumer. + * @param partitions the partitions. + * @param initial true if this is the initial assignment. + */ + default void onPartitionsAssigned(String bindingName, Consumer consumer, Collection partitions, + boolean initial) { + + } + +} +---- +==== + +You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener. + = Appendices [appendix] [[building]] diff --git a/docs/src/main/asciidoc/dlq.adoc b/docs/src/main/asciidoc/dlq.adoc index cd03abfd5..ea05d2a28 100644 --- a/docs/src/main/asciidoc/dlq.adoc +++ b/docs/src/main/asciidoc/dlq.adoc @@ -1,5 +1,5 @@ [[kafka-dlq-processing]] -== Dead-Letter Topic Processing +=== Dead-Letter Topic Processing Because you cannot anticipate how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. If the reason for the dead-lettering is transient, you may wish to route the messages back to the original topic. diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index ef3f5532a..b41f4328f 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -1,4 +1,6 @@ -== Usage +== Kafka Streams Binder + +=== Usage For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates: @@ -11,7 +13,8 @@ Maven coordinates: ---- -== Kafka Streams Binder Overview + +=== Overview Spring Cloud Stream's Apache Kafka support also includes a binder implementation designed explicitly for Apache Kafka Streams binding. With this native integration, a Spring Cloud Stream "processor" application can directly use the @@ -32,7 +35,7 @@ As noted early-on, Kafka Streams support in Spring Cloud Stream is strictly only A model in which the messages read from an inbound topic, business processing can be applied, and the transformed messages can be written to an outbound topic. It can also be used in Processor applications with a no-outbound destination. -=== Streams DSL +==== Streams DSL This application consumes data from a Kafka topic (e.g., `words`), computes word count for each unique word in a 5 seconds time window, and the computed results are sent to a downstream topic (e.g., `counts`) for further processing. @@ -75,13 +78,13 @@ KStream objects. As a developer, you can exclusively focus on the business aspec required in the processor. Setting up the Streams DSL specific configuration required by the Kafka Streams infrastructure is automatically handled by the framework. -== Configuration Options +=== Configuration Options This section contains the configuration options used by the Kafka Streams binder. For common configuration options and properties pertaining to binder, refer to the <>. -=== Kafka Streams Properties +==== Kafka Streams Properties The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` @@ -172,7 +175,7 @@ Default: `earliest`. Note: Using `resetOffsets` on the consumer does not have any effect on Kafka Streams binder. Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand. -=== TimeWindow properties: +==== TimeWindow properties: Windowing is an important concept in stream processing applications. Following properties are available to configure time-window computations. @@ -187,14 +190,14 @@ spring.cloud.stream.kafka.streams.timeWindow.advanceBy:: + Default: `none`. -== Multiple Input Bindings +=== Multiple Input Bindings For use cases that requires multiple incoming KStream objects or a combination of KStream and KTable objects, the Kafka Streams binder provides multiple bindings support. Let's see it in action. -=== Multiple Input Bindings as a Sink +==== Multiple Input Bindings as a Sink [source] ---- @@ -261,7 +264,7 @@ interface KStreamKTableBinding extends KafkaStreamsProcessor { ---- -== Multiple Output Bindings (aka Branching) +=== Multiple Output Bindings (aka Branching) Kafka Streams allow outbound data to be split into multiple topics based on some predicates. The Kafka Streams binder provides support for this feature without compromising the programming model exposed through `StreamListener` in the end user application. @@ -347,7 +350,7 @@ spring.cloud.stream.bindings.input: headerMode: raw ---- -== Record Value Conversion +=== Record Value Conversion Kafka Streams binder can marshal producer/consumer values based on a content type and the converters provided out of the box in Spring Cloud Stream. @@ -359,7 +362,7 @@ would like to continue using that for inbound and outbound conversions. Both the options are supported in the Kafka Streams binder implementation. See below for more details. -==== Outbound serialization +===== Outbound serialization If native encoding is disabled (which is the default), then the framework will convert the message using the contentType set by the user (otherwise, the default `application/json` will be applied). It will ignore any SerDe set on the outbound @@ -448,7 +451,7 @@ spring.cloud.stream.bindings.output2.contentType: application/java-serialzied-ob spring.cloud.stream.bindings.output3.contentType: application/octet-stream ---- -==== Inbound Deserialization +===== Inbound Deserialization Similar rules apply to data deserialization on the inbound. @@ -503,7 +506,7 @@ As in the case of KStream branching on the outbound, the benefit of setting valu multiple input bindings (multiple KStreams object) and they all require separate value SerDe's, then you can configure them individually. If you use the common configuration approach, then this feature won't be applicable. -== Error Handling +=== Error Handling Apache Kafka Streams provide the capability for natively handling exceptions from deserialization errors. For details on this support, please see https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+deserialization+exception+handlers[this] @@ -544,7 +547,7 @@ that if there are multiple `StreamListener` methods in the same application, thi * The exception handling for deserialization works consistently with native deserialization and framework provided message conversion. -=== Handling Non-Deserialization Exceptions +==== Handling Non-Deserialization Exceptions For general error handling in Kafka Streams binder, it is up to the end user applications to handle application level errors. As a side effect of providing a DLQ for deserialization exception handlers, Kafka Streams binder provides a way to get @@ -594,7 +597,7 @@ public KStream process(KStream input) { } ---- -== State Store +=== State Store State store is created automatically by Kafka Streams when the DSL is used. When processor API is used, you need to register a state store manually. In order to do so, you can use `KafkaStreamsStateStore` annotation. @@ -626,7 +629,7 @@ Processor() { } ---- -== Interactive Queries +=== Interactive Queries As part of the public Kafka Streams binder API, we expose a class called `InteractiveQueryService`. You can access this as a Spring bean in your application. An easy way to get access to this bean from your application is to "autowire" the bean. @@ -671,7 +674,7 @@ else { } ---- -== Accessing the underlying KafkaStreams object +=== Accessing the underlying KafkaStreams object `StreamBuilderFactoryBean` from spring-kafka that is responsible for constructing the `KafkaStreams` object can be accessed programmatically. Each `StreamBuilderFactoryBean` is registered as `stream-builder` and appended with the `StreamListener` method name. @@ -685,7 +688,7 @@ StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-b KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams(); ---- -== State Cleanup +=== State Cleanup By default, the `Kafkastreams.cleanup()` method is called when the binding is stopped. See https://docs.spring.io/spring-kafka/reference/html/_reference.html#_configuration[the Spring Kafka documentation]. diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 24543600d..667ddc393 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -5,7 +5,9 @@ It contains information about its design, usage, and configuration options, as w In addition, this guide explains the Kafka Streams binding capabilities of Spring Cloud Stream. -- -== Usage +== Apache Kafka Binder + +=== Usage To use Apache Kafka binder, you need to add `spring-cloud-stream-binder-kafka` as a dependency to your Spring Cloud Stream application, as shown in the following example for Maven: @@ -27,7 +29,7 @@ Alternatively, you can also use the Spring Cloud Stream Kafka Starter, as shown ---- -== Apache Kafka Binder Overview +=== Overview The following image shows a simplified diagram of how the Apache Kafka binder operates: @@ -43,13 +45,13 @@ This client can communicate with older brokers (see the Kafka documentation), bu 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. -== Configuration Options +=== Configuration Options This section contains the configuration options used by the Apache Kafka binder. For common configuration options and properties pertaining to binder, see the <>. -=== Kafka Binder Properties +==== Kafka Binder Properties spring.cloud.stream.kafka.binder.brokers:: A list of brokers to which the Kafka binder connects. @@ -138,7 +140,7 @@ Use this, for example, if you wish to customize the trusted packages in a `Defau Default: none. [[kafka-consumer-properties]] -=== Kafka Consumer Properties +==== Kafka Consumer Properties The following properties are available for Kafka consumers only and must be prefixed with `spring.cloud.stream.kafka.bindings..consumer.`. @@ -254,7 +256,7 @@ This can be configured using the `configuration` property above. Default: `false` [[kafka-producer-properties]] -=== Kafka Producer Properties +==== Kafka Producer Properties The following properties are available for Kafka producers only and must be prefixed with `spring.cloud.stream.kafka.bindings..producer.`. @@ -315,11 +317,11 @@ If a topic already exists with a smaller partition count and `autoAddPartitions` If a topic already exists with a smaller partition count and `autoAddPartitions` is enabled, new partitions are added. If a topic already exists with a larger number of partitions than the maximum of (`minPartitionCount` or `partitionCount`), the existing partition count is used. -=== Usage examples +==== Usage examples In this section, we show the use of the preceding properties for specific scenarios. -==== Example: Setting `autoCommitOffset` to `false` and Relying on Manual Acking +===== Example: Setting `autoCommitOffset` to `false` and Relying on Manual Acking This example illustrates how one may manually acknowledge offsets in a consumer application. @@ -347,7 +349,7 @@ public class ManuallyAcknowdledgingConsumer { } ---- -==== 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 http://kafka.apache.org/090/documentation.html#security_configclients[Apache Kafka Documentation] as well as the Kafka 0.9 http://docs.confluent.io/2.0.0/kafka/security.html[security guidelines from the Confluent documentation]. @@ -366,7 +368,7 @@ When using Kerberos, follow the instructions in the http://kafka.apache.org/090/ 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: @@ -379,7 +381,7 @@ The following example shows how to launch a Spring Cloud Stream application with --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. @@ -436,7 +438,7 @@ 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. [[pause-resume]] -==== Example: Pausing and Resuming the Consumer +===== 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 adding the `Consumer` as a parameter to your `@StreamListener`. @@ -476,7 +478,7 @@ public class Application { ---- [[kafka-error-channels]] -== Error Channels +=== Error Channels Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel. See <> for more information. @@ -490,7 +492,7 @@ There is no automatic handling of producer exceptions (such as sending to a <