diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc index 3b0752c1f..e5b6a3aaa 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc @@ -1,6 +1,6 @@ == Usage -For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following +For using the Kafka Streams binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates: [source,xml] @@ -13,26 +13,26 @@ Maven coordinates: == Kafka Streams Binder 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 +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 https://kafka.apache.org/documentation/streams/developer-guide[Apache Kafka Streams] APIs in the core business logic. -Kafka Streams binder implementation builds on the foundation provided by the http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafka Streams in Spring Kafka] +Kafka Streams binder implementation builds on the foundation provided by the http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafka Streams in Spring Kafka] project. -As part of this native integration, the high-level https://docs.confluent.io/current/streams/developer-guide/dsl-api.html[Streams DSL] +As part of this native integration, the high-level https://docs.confluent.io/current/streams/developer-guide/dsl-api.html[Streams DSL] provided by the Kafka Streams API is available for use in the business logic, too. -An early version of the https://docs.confluent.io/current/streams/developer-guide/processor-api.html[Processor API] +An early version of the https://docs.confluent.io/current/streams/developer-guide/processor-api.html[Processor API] support is available as well. -As noted early-on, Kafka Streams support in Spring Cloud Stream strictly only available for use in the Processor model. -A model in which the messages read from an inbound topic, business processing can be applied, and the transformed messages +As noted early-on, Kafka Streams support in Spring Cloud Stream strictly only available for use in the Processor model. +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 -This application consumes data from a Kafka topic (e.g., `words`), computes word count for each unique word in a 5 seconds +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. [source] @@ -65,12 +65,12 @@ Once built as a uber-jar (e.g., `wordcount-processor.jar`), you can run the abov java -jar wordcount-processor.jar --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts ---- -This application will consume messages from the Kafka topic `words` and the computed results are published to an output +This application will consume messages from the Kafka topic `words` and the computed results are published to an output 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 +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 is automatically handled by the framework. == Configuration Options @@ -81,7 +81,7 @@ For common configuration options and properties pertaining to binder, refer to t === Kafka Streams Properties -The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` +The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` literal. configuration:: @@ -96,7 +96,7 @@ spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.a spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000 ---- -For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs in +For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs in Apache Kafka Streams docs. brokers:: @@ -119,7 +119,7 @@ applicationId:: + Default: `default` -The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..producer.` +The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..producer.` literal. keySerde:: @@ -135,7 +135,7 @@ useNativeEncoding:: + Default: `false`. -The following properties are _only_ available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..consumer.` +The following properties are _only_ available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..consumer.` literal. keySerde:: @@ -176,8 +176,8 @@ Default: `none`. == 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. +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. @@ -206,11 +206,11 @@ interface KStreamKTableBinding { ---- -In the above example, the application is written as a sink, i.e. there are no output bindings and the application has to -decide concerning downstream processing. When you write applications in this style, you might want to send the information +In the above example, the application is written as a sink, i.e. there are no output bindings and the application has to +decide concerning downstream processing. When you write applications in this style, you might want to send the information downstream or store them in a state store (See below for Queryable State Stores). -In the case of incoming KTable, if you want to materialize the computations to a state store, you have to express it +In the case of incoming KTable, if you want to materialize the computations to a state store, you have to express it through the following property. [source] @@ -244,13 +244,13 @@ interface KStreamKTableBinding extends KafkaStreamsProcessor { == 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 +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. -You can write the application in the usual way as demonstrated above in the word count example. However, when using the -branching feature, you are required to do a few things. First, you need to make sure that your return type is `KStream[]` -instead of a regular `KStream`. Second, you need to use the `SendTo` annotation containing the output bindings in the order -(see example below). For each of these output bindings, you need to configure destination, content-type etc., complying with +You can write the application in the usual way as demonstrated above in the word count example. However, when using the +branching feature, you are required to do a few things. First, you need to make sure that your return type is `KStream[]` +instead of a regular `KStream`. Second, you need to use the `SendTo` annotation containing the output bindings in the order +(see example below). For each of these output bindings, you need to configure destination, content-type etc., complying with the standard Spring Cloud Stream expectations. Here is an example: @@ -330,21 +330,21 @@ spring.cloud.stream.bindings.input: == Message Conversion -Similar to message-channel based binder applications, the Kafka Streams binder adapts to the out-of-the-box content-type +Similar to message-channel based binder applications, the Kafka Streams binder adapts to the out-of-the-box content-type conversions without any compromise. It is typical for Kafka Streams operations to know the type of SerDe’s used to transform the key and value correctly. -Therefore, it may be more natural to rely on the SerDe facilities provided by the Apache Kafka Streams library itself at +Therefore, it may be more natural to rely on the SerDe facilities provided by the Apache Kafka Streams library itself at the inbound and outbound conversions rather than using the content-type conversions offered by the framework. -On the other hand, you might be already familiar with the content-type conversion patterns provided by the framework, and +On the other hand, you might be already familiar with the content-type conversion patterns provided by the framework, and that, you'd like to continue using for inbound and outbound conversions. -Both the options are supported in the Kafka Streams binder implementation. +Both the options are supported in the Kafka Streams binder implementation. ==== 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 +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 in this case for outbound serialization. Here is the property to set the contentType on the outbound. @@ -361,7 +361,7 @@ Here is the property to enable native encoding. spring.cloud.stream.bindings.output.nativeEncoding: true ---- -If native encoding is enabled on the output binding (user has to enable it as above explicitly), then the framework will +If native encoding is enabled on the output binding (user has to enable it as above explicitly), then the framework will skip any form of automatic message conversion on the outbound. In that case, it will switch to the Serde set by the user. The `valueSerde` property set on the actual output binding will be used. Here is an example. @@ -372,7 +372,7 @@ spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde: org.apach If this property is not set, then it will use the "default" SerDe: `spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde`. It is worth to mention that Kafka Streams binder does not serialize the keys on outbound - it simply relies on Kafka itself. -Therefore, you either have to specify the `keySerde` property on the binding or it will default to the application-wide common +Therefore, you either have to specify the `keySerde` property on the binding or it will default to the application-wide common `keySerde`. Binding level key serde: @@ -418,9 +418,9 @@ spring.cloud.stream.kafka.streams.bindings.output2.producer.valueSerde=StringSer spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=JsonSerde ---- -Then if you have `SendTo` like this, @SendTo({"output1", "output2", "output3"}), the `KStream[]` from the branches are -applied with proper SerDe objects as defined above. If you are not enabling `nativeEncoding`, you can then set different -contentType values on the output bindings as below. In that case, the framework will use the appropriate message converter +Then if you have `SendTo` like this, @SendTo({"output1", "output2", "output3"}), the `KStream[]` from the branches are +applied with proper SerDe objects as defined above. If you are not enabling `nativeEncoding`, you can then set different +contentType values on the output bindings as below. In that case, the framework will use the appropriate message converter to convert the messages before sending to Kafka. [source] @@ -434,8 +434,8 @@ spring.cloud.stream.bindings.output3.contentType: application/octet-stream Similar rules apply to data deserialization on the inbound. -If native decoding 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 inbound +If native decoding 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 inbound in this case for inbound deserialization. Here is the property to set the contentType on the inbound. @@ -452,8 +452,8 @@ Here is the property to enable native decoding. spring.cloud.stream.bindings.input.nativeDecoding: true ---- -If native decoding is enabled on the input binding (user has to enable it as above explicitly), then the framework will -skip doing any message conversion on the inbound. In that case, it will switch to the SerDe set by the user. The `valueSerde` +If native decoding is enabled on the input binding (user has to enable it as above explicitly), then the framework will +skip doing any message conversion on the inbound. In that case, it will switch to the SerDe set by the user. The `valueSerde` property set on the actual output binding will be used. Here is an example. [source] @@ -464,7 +464,7 @@ spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde: org.apache If this property is not set, it will use the default SerDe: `spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde`. It is worth to mention that Kafka Streams binder does not deserialize the keys on inbound - it simply relies on Kafka itself. -Therefore, you either have to specify the `keySerde` property on the binding or it will default to the application-wide common +Therefore, you either have to specify the `keySerde` property on the binding or it will default to the application-wide common `keySerde`. Binding level key serde: @@ -481,8 +481,8 @@ Common Key serde: spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde ---- -As in the case of KStream branching on the outbound, the benefit of setting value SerDe per binding is that if you have -multiple input bindings (multiple KStreams object) and they all require separate value SerDe's, then you can configure +As in the case of KStream branching on the outbound, the benefit of setting value SerDe per binding is that if you have +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 @@ -490,7 +490,7 @@ them individually. If you use the common configuration approach, then this featu 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] Out of the box, Apache Kafka Streams provide two kinds of deserialization exception handlers - `logAndContinue` and `logAndFail`. -As the name indicates, the former will log the error and continue processing the next records and the latter will log the +As the name indicates, the former will log the error and continue processing the next records and the latter will log the error and fail. `LogAndFail` is the default deserialization exception handler. === Handling Deserialization Exceptions @@ -502,7 +502,7 @@ Kafka Streams binder supports a selection of exception handlers through the foll spring.cloud.stream.kafka.streams.binder.serdeError: logAndContinue ---- -In addition to the above two deserialization exception handlers, the binder also provides a third one for sending the erroneous +In addition to the above two deserialization exception handlers, the binder also provides a third one for sending the erroneous records (poison pills) to a DLQ topic. Here is how you enable this DLQ exception handler. [source] @@ -516,27 +516,27 @@ When the above property is set, all the deserialization error records are automa spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: foo-dlq ---- -If this is set, then the error records are sent to the topic `foo-dlq`. If this is not set, then it will create a DLQ +If this is set, then the error records are sent to the topic `foo-dlq`. If this is not set, then it will create a DLQ topic with the name `error..`. A couple of things to keep in mind when using the exception handling feature in Kafka Streams binder. -* The property `spring.cloud.stream.kafka.streams.binder.serdeError` is applicable for the entire application. This implies +* The property `spring.cloud.stream.kafka.streams.binder.serdeError` is applicable for the entire application. This implies that if there are multiple `StreamListener` methods in the same application, this property is applied to all of them. -* The exception handling for deserialization works consistently with native deserialization and framework provided message +* The exception handling for deserialization works consistently with native deserialization and framework provided message conversion. === 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 +As a side effect of providing a DLQ for deserialization exception handlers, Kafka Streams binder provides a way to get access to the DLQ sending bean directly from your application. Once you get access to that bean, you can programmatically send any exception records from your application to the DLQ. -It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn't natively support error -handling yet. +It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn't natively support error +handling yet. -However, when you use the low-level Processor API in your application, there are options to control this behavior. See +However, when you use the low-level Processor API in your application, there are options to control this behavior. See below. [source] @@ -665,4 +665,10 @@ Following is an example and it assumes the `StreamListener` method is named as ` ---- StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams(); ----- \ No newline at end of file +---- + +== 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]. +To modify this behavior simply add a single `CleanupConfig` `@Bean` (configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index e30a77e79..b3fdaf8fe 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -25,6 +25,7 @@ import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; import org.apache.kafka.streams.errors.LogAndFailExceptionHandler; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; @@ -37,6 +38,7 @@ import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.CleanupConfig; import org.springframework.util.ObjectUtils; /** @@ -99,10 +101,12 @@ public class KafkaStreamsBinderSupportAutoConfiguration { KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, KStreamStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter, Collection streamListenerResultAdapters, - KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + ObjectProvider cleanupConfig) { return new KafkaStreamsStreamListenerSetupMethodOrchestrator(bindingServiceProperties, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, kafkaStreamsBindingInformationCatalogue, - kafkaStreamListenerParameterAdapter, streamListenerResultAdapters, binderConfigurationProperties); + kafkaStreamListenerParameterAdapter, streamListenerResultAdapters, binderConfigurationProperties, + cleanupConfig.getIfUnique()); } @Bean diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index eceaed414..fe7c7b83a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -62,6 +62,7 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.MethodParameter; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -84,6 +85,7 @@ import org.springframework.util.StringUtils; * * @author Soby Chacko * @author Lei Chen + * @author Gary Russell */ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator, ApplicationContextAware { @@ -105,6 +107,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; + private final CleanupConfig cleanupConfig; + private ConfigurableApplicationContext applicationContext; KafkaStreamsStreamListenerSetupMethodOrchestrator(BindingServiceProperties bindingServiceProperties, @@ -113,7 +117,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, StreamListenerParameterAdapter streamListenerParameterAdapter, Collection streamListenerResultAdapters, - KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + CleanupConfig cleanupConfig) { this.bindingServiceProperties = bindingServiceProperties; this.kafkaStreamsExtendedBindingProperties = kafkaStreamsExtendedBindingProperties; this.keyValueSerdeResolver = keyValueSerdeResolver; @@ -121,6 +126,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene this.streamListenerParameterAdapter = streamListenerParameterAdapter; this.streamListenerResultAdapters = streamListenerResultAdapters; this.binderConfigurationProperties = binderConfigurationProperties; + this.cleanupConfig = cleanupConfig; } @Override @@ -382,13 +388,6 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene private StreamsConfig buildStreamsBuilderAndRetrieveConfig(Method method, ApplicationContext applicationContext, BindingProperties bindingProperties) { ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory(); - StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean(); - streamsBuilder.setAutoStartup(false); - BeanDefinition streamsBuilderBeanDefinition = - BeanDefinitionBuilder.genericBeanDefinition((Class) streamsBuilder.getClass(), () -> streamsBuilder) - .getRawBeanDefinition(); - ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + method.getName(), streamsBuilderBeanDefinition); - StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class); String group = bindingProperties.getGroup(); if (!StringUtils.hasText(group)) { group = binderConfigurationProperties.getApplicationId(); @@ -415,12 +414,20 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene return super.getConfiguredInstance(key, clazz); } }; + StreamsBuilderFactoryBean streamsBuilder = this.cleanupConfig == null + ? new StreamsBuilderFactoryBean(streamsConfig) + : new StreamsBuilderFactoryBean(streamsConfig, this.cleanupConfig); + streamsBuilder.setAutoStartup(false); + BeanDefinition streamsBuilderBeanDefinition = + BeanDefinitionBuilder.genericBeanDefinition((Class) streamsBuilder.getClass(), () -> streamsBuilder) + .getRawBeanDefinition(); + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + method.getName(), streamsBuilderBeanDefinition); + StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class); BeanDefinition streamsConfigBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition((Class) streamsConfig.getClass(), () -> streamsConfig) .getRawBeanDefinition(); ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("streamsConfig-" + method.getName(), streamsConfigBeanDefinition); - streamsBuilder.setStreamsConfig(streamsConfig); methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX); return streamsConfig; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderWordCountIntegrationTests.java index 001cd3f9f..9dc2186e6 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderWordCountIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -48,6 +48,9 @@ import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; @@ -110,8 +113,12 @@ public class KafkaStreamsBinderWordCountIntegrationTests { KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams(); ReadOnlyWindowStore store = kafkaStreams.store("foo-WordCounts", QueryableStoreTypes.windowStore()); assertThat(store).isNotNull(); - - } finally { + CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean, "cleanupConfig", + CleanupConfig.class); + assertThat(cleanup.cleanupOnStart()).isTrue(); + assertThat(cleanup.cleanupOnStop()).isFalse(); + } + finally { context.close(); } } @@ -139,8 +146,6 @@ public class KafkaStreamsBinderWordCountIntegrationTests { public KStream process(@Input("input") KStream input) { input.map((k,v) -> { - System.out.println(k); - System.out.println(v); return new KeyValue<>(k,v); }); return input @@ -153,6 +158,11 @@ public class KafkaStreamsBinderWordCountIntegrationTests { .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); } + @Bean + public CleanupConfig cleanupConfig() { + return new CleanupConfig(true, false); + } + } static class WordCount { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java index 8e415a5af..5b14dd1f4 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -38,9 +38,12 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -92,7 +95,15 @@ public class KafkastreamsBinderPojoInputStringOutputIntegrationTests { "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { receiveAndValidateFoo(context); - } finally { + //Assertions on StreamBuilderFactoryBean + StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", + StreamsBuilderFactoryBean.class); + CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean, "cleanupConfig", + CleanupConfig.class); + assertThat(cleanup.cleanupOnStart()).isFalse(); + assertThat(cleanup.cleanupOnStop()).isTrue(); + } + finally { context.close(); } }