From fefd9a3bd6965961fb042658a7af0d46bdc04a38 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 7 Nov 2019 18:07:27 -0500 Subject: [PATCH] Cleanup in Kafka Streams docs --- docs/src/main/asciidoc/kafka-streams.adoc | 165 ++++++++++------------ 1 file changed, 78 insertions(+), 87 deletions(-) diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 9ea1402ae..616de99da 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -111,7 +111,7 @@ public class WordCountProcessorApplication { } ---- -Here again, this is a complete Spring Boot application. The difference here from the first application, though, the bean method is of type `java.util.function.Function`. +Here again, this is a complete Spring Boot application. The difference here from the first application is that the bean method is of type `java.util.function.Function`. The first parameterized type for the `Function` is for the input `KStream` and the second one is for the output. In the method body, a lambda expression is provided that is of type `Function` and as implementation, the actual business logic is given. Similar to the previously discussed Consumer based application, the input binding here is named as `process-in-0` by default. For the output, the binding name is automatically also set to `process-out-0`. @@ -120,7 +120,7 @@ Once built as a uber-jar (e.g., `wordcount-processor.jar`), you can run the abov [source] ---- -java -jar wordcount-processor.jar --spring.cloud.stream.bindings.process-in-0.destination=words --spring.cloud.stream.bindings.process-out-0.destination=counts +java -jar wordcount-processor.jar --spring.cloud.stream.bindings.process-in-0.destination=words --spring.cloud.stream.bindings.process-out-0.destination=counts ---- This application will consume messages from the Kafka topic `words` and the computed results are published to an output @@ -128,7 +128,7 @@ topic `counts`. Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are automatically bound as KStream objects. As a developer, you can exclusively focus on the business aspects of the code, i.e. writing the logic -required in the processor. Setting up the Streams DSL specific configuration required by the Kafka Streams infrastructure +required in the processor. Setting up Kafka Streams specific configuration required by the Kafka Streams infrastructure is automatically handled by the framework. The two examples we saw above have a single `KStream` input binding. In both cases, the bindings received the records from a single topic. @@ -165,15 +165,14 @@ public BiFunction, KTable, KStream, KTable> process() { } ---- +====== Beyond two inputs + What if you have more than two inputs? There are situations in which you need more than two inputs. In that case, the binder allows you to chain partial functions. In functional programming jargon, this technique is generally known as currying. @@ -264,8 +265,9 @@ public Function, KStream[]> process() { The programming model remains the same, however the outbound parameterized type is `KStream[]`. The default output binding names are `process-out-0`, `process-out-1`, `process-out-2` respectively. +The reason why the binder generates three output bindings is because it detects the length of the returned `KStream` array. -===== Function based Programming Styles for Kafka Streams +===== Summary of Function based Programming Styles for Kafka Streams In summary, the following table shows the various options that can be used in the functional paradigm. @@ -395,26 +397,26 @@ Finally, here is the `StreamListener` equivalent of the application with three i ... ... @StreamListener - @SendTo("output") - public KStream process( - @Input("input-1") KStream ordersStream, - @Input("input-"2) GlobalKTable customers, - @Input("input-3") GlobalKTable products) { + @SendTo("output") + public KStream process( + @Input("input-1") KStream ordersStream, + @Input("input-"2) GlobalKTable customers, + @Input("input-3") GlobalKTable products) { - KStream customerOrdersStream = ordersStream.join( - customers, (orderId, order) -> order.getCustomerId(), - (order, customer) -> new CustomerOrder(customer, order)); + KStream customerOrdersStream = ordersStream.join( + customers, (orderId, order) -> order.getCustomerId(), + (order, customer) -> new CustomerOrder(customer, order)); - return customerOrdersStream.join(products, - (orderId, customerOrder) -> customerOrder.productId(), - (customerOrder, product) -> { - EnrichedOrder enrichedOrder = new EnrichedOrder(); - enrichedOrder.setProduct(product); - enrichedOrder.setCustomer(customerOrder.customer); - enrichedOrder.setOrder(customerOrder.order); - return enrichedOrder; - }); - } + return customerOrdersStream.join(products, + (orderId, customerOrder) -> customerOrder.productId(), + (customerOrder, product) -> { + EnrichedOrder enrichedOrder = new EnrichedOrder(); + enrichedOrder.setProduct(product); + enrichedOrder.setCustomer(customerOrder.customer); + enrichedOrder.setOrder(customerOrder.order); + return enrichedOrder; + }); + } interface CustomGlobalKTableProcessor { @@ -436,7 +438,7 @@ Finally, here is the `StreamListener` equivalent of the application with three i You might notice that the above two examples are even more verbose since in addition to provide `EnableBinding`, you also need to write your own custom binding interface as well. Using the functional model, you can avoid all those ceremonial details. -Before we move on looking at the general programming model offered by Kafka Streams binder, here is the `StreamListener` version of multiple output bindings. +Before we move on from looking at the general programming model offered by Kafka Streams binder, here is the `StreamListener` version of multiple output bindings. [source] ---- @@ -515,12 +517,12 @@ public java.util.function.BiFunction, KTable, KTable, KStream ---- -Spring Cloud Stream Binder Kafka Streams provides a health indicator to check the state of the underlying Kafka threads. +Spring Cloud Stream Kafka Streams Binder provides a health indicator to check the state of the underlying streams threads. Spring Cloud Stream defines a property `management.health.binders.enabled` to enable the health indicator. See the https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_health_indicator[Spring Cloud Stream documentation]. -The health indicator provides the following details for each Kafka threads: +The health indicator provides the following details for each stream thread's metadata: * Thread name * Thread state: `CREATED`, `RUNNING`, `PARTITIONS_REVOKED`, `PARTITIONS_ASSIGNED`, `PENDING_SHUTDOWN` or `DEAD` @@ -1033,7 +1037,7 @@ For example, if the application ID of the first processor is `processor-1`, then You can either programmatically access the Micrometer `MeterRegistry` in the application and then iterate through the available gauges or use Spring Boot actuator to access the metrics through a REST endpoint. When accessing through the Boot actuator endpoint, make sure to add `metrics` to the property `management.endpoints.web.exposure.include`. -Then you can access `/acutator/metrics` to get a list of all the available metrics which then can be individually accessed through the same URL (`/actuator/metrics/`). +Then you can access `/acutator/metrics` to get a list of all the available metrics which then can be individually accessed through the same URI (`/actuator/metrics/`). Anything beyond the info level metrics available through `KafkaStreams#metrics()`, (for e.g. the debugging level metrics) are still only available through JMX after you set the `metrics.recording.level` to `DEBUG`. Kafka Streams, by default, set this level to `INFO`. @@ -1125,21 +1129,21 @@ Here is the output binding destination: spring.cloud.stream.bindings.process-out-0.destination: outputTopic ``` -If the topic `outputTopic` has 4 partitions, if you don't provide a partitioning strategy, Kafka Streams will use default partitioning strategy which may or may not work depending on the particular use case. -Let's say, you want to send any key that matches to `foo` to partition 0, `bar` to partion 1, `baz` to partition 2, and everything else to partition 3. +If the topic `outputTopic` has 4 partitions, if you don't provide a partitioning strategy, Kafka Streams will use default partitioning strategy which may not be the outcome you want depending on the particular use case. +Let's say, you want to send any key that matches to `spring` to partition 0, `cloud` to partion 1, `stream` to partition 2, and everything else to partition 3. This is what you need to do in the application. ``` @Bean public StreamPartitioner streamPartitioner() { return (t, k, v, n) -> { - if (k.equals("foo")) { + if (k.equals("spring")) { return 0; } - else if (k.equals("bar")) { + else if (k.equals("cloud")) { return 1; } - else if (k.equals("baz")) { + else if (k.equals("stream")) { return 2; } else { @@ -1149,7 +1153,7 @@ public StreamPartitioner streamPartitioner() { } ``` -This is a rudimentary implementation, however, you have access to the key and value of the record, the topic name and the total number of partitions. +This is a rudimentary implementation, however, you have access to the key/value of the record, the topic name and the total number of partitions. Therefore, you can implement complex partiioning strategies if need be. You also need to provide this bean name along with the application configuration. @@ -1163,8 +1167,8 @@ Each output topic in the application needs to be configured separately like this === StreamsBuilderFactoryBean customizer It is often required to customize the `StreamsBuilderFactoryBean` that creates the `KafkaStreams` objects. -Based on the underlying support provided by Spring Kafka, the binder allows you to customize the `StreamsBuilderFactoryBean` in two ways. -One, you can use the `StreamsBuilderFactoryBeanCustomizer` to customize the `StreamsBuilderFactoryBean` itself. +Based on the underlying support provided by Spring Kafka, the binder allows you to customize the `StreamsBuilderFactoryBean`. +You can use the `StreamsBuilderFactoryBeanCustomizer` to customize the `StreamsBuilderFactoryBean` itself. Then, once you get access to the `StreamsBuilderFactoryBean` through this customizer, you can customize the corresponding `KafkaStreams` using `KafkaStreamsCustomzier`. Both of these customizers are part of the Spring for Apache Kafka project. @@ -1205,7 +1209,7 @@ public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() `KafkaStreamsCustomizer` will be called by the `StreamsBuilderFactoryBeabn` right before the underlying `KafkaStreams` gets started. There can only be one `StreamsBuilderFactoryBeanCustomizer` in the entire application. -Then how do we account for multiple Kafka Streams processors as each of them are backed up by `StreamsBuilderFactoryBeabn`. +Then how do we account for multiple Kafka Streams processors as each of them are backed up by individual `StreamsBuilderFactoryBean` objects? In that case, if the customization needs to be different for those processors, then the application needs to apply some filter based on the application ID. For e.g, @@ -1231,10 +1235,10 @@ public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer() === Timestamp extractor -Kafka Streams allows you to control the the processing of the consumer records based on various notions of timestamp. +Kafka Streams allows you to control the processing of the consumer records based on various notions of timestamp. By default, Kafka Streams extracts the timestamp metadata embedded in the consumer record. You can change this default behavior by providing a different `TimestampExtractor` implementation per input binding. -Here are some details on how to do so. +Here are some details on how that can be done. ``` @Bean @@ -1293,7 +1297,7 @@ spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar ``` -Things become a bit more complex if you have the same application as above, but is dealing with two different Kafka clusters, for e.g. the regular process is acting upon both Kafka cluster 1 and cluster 2(receiving data from cluster-1 and sending to cluster-2) and the Kafka Streams processor is acting upon Kafka cluster 2. +Things become a bit more complex if you have the same application as above, but is dealing with two different Kafka clusters, for e.g. the regular `process` is acting upon both Kafka cluster 1 and cluster 2 (receiving data from cluster-1 and sending to cluster-2) and the Kafka Streams processor is acting upon Kafka cluster 2. Then you have to use the https://cloud.spring.io/spring-cloud-stream/reference/html/spring-cloud-stream.html#multiple-binders[multibinder] facilities provided by Spring Cloud Stream. Here is how your configuration may change in that scenario. @@ -1307,7 +1311,6 @@ spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams spring.cloud.stream.binders.kafka3.type: kstream spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster - spring.cloud.stream.function.definition=process;kstreamProcess # From cluster 1 to cluster 2 with regular process function @@ -1324,7 +1327,7 @@ spring.cloud.stream.bindings.kstreamProcess-out-0.binder=kafka3 ``` Pay attention to the above configuration. -We have two kinds of binders, but 3 binders all in all, first one is the regular Kafka binder based on cluster 1 (`kafka1`), then another Kafka binder based on cluster 2 (`kafka2`) and finally the kstream on (`kafka3`) +We have two kinds of binders, but 3 binders all in all, first one is the regular Kafka binder based on cluster 1 (`kafka1`), then another Kafka binder based on cluster 2 (`kafka2`) and finally the `kstream` one (`kafka3`). The first processor in the application receives data from `kafka1` and publishes to `kafka2` where both binders are based on regular Kafka binder but differnt clusters. The second processor, which is a Kafka Streams processor consumes data from `kafka3` which is the same cluster as `kafka2`, but a different binder type. @@ -1342,7 +1345,9 @@ public Function, } ``` -then, this has to be configured in a multi binder scenario as the following: +then, this has to be configured in a multi binder scenario as the following. +Please note that this is only needed if you have a true multi-binder scenario where there are multiple processors dealing with multiple clusters within a single application. +In that case, the binders need to be explicitly provided with the bindings to distinguish from other processor's binder types and clusters. ``` spring.cloud.stream.binders.kafka1.type: kstream @@ -1465,17 +1470,3 @@ 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. - -=== 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. -If your `StreamListener` method is named as `process` for example, the stream builder bean is named as `stream-builder-process`. -Since this is a factory bean, it should be accessed by prepending an ampersand (`&`) when accessing it programmatically. -Following is an example and it assumes the `StreamListener` method is named as `process` - -[source] ----- -StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); - KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams(); ----- \ No newline at end of file