diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index dc966e035..c87908840 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -355,7 +355,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application private Consumed getConsumed(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { TimestampExtractor timestampExtractor = null; - if (kafkaStreamsConsumerProperties.getTimestampExtractorBeanName() != null) { + if (!StringUtils.isEmpty(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName())) { timestampExtractor = applicationContext.getBean(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName(), TimestampExtractor.class); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/EncodingDecodingBindAdviceHandler.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/EncodingDecodingBindAdviceHandler.java new file mode 100644 index 000000000..91e50adee --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/EncodingDecodingBindAdviceHandler.java @@ -0,0 +1,72 @@ +/* + * Copyright 2019-2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import org.springframework.boot.context.properties.ConfigurationPropertiesBindHandlerAdvisor; +import org.springframework.boot.context.properties.bind.AbstractBindHandler; +import org.springframework.boot.context.properties.bind.BindContext; +import org.springframework.boot.context.properties.bind.BindHandler; +import org.springframework.boot.context.properties.bind.BindResult; +import org.springframework.boot.context.properties.bind.Bindable; +import org.springframework.boot.context.properties.source.ConfigurationPropertyName; + +/** + * {@link ConfigurationPropertiesBindHandlerAdvisor} to detect nativeEncoding/Decoding settings + * provided by the application explicitly. + * + * @author Soby Chacko + * @since 3.0.0 + */ +public class EncodingDecodingBindAdviceHandler implements ConfigurationPropertiesBindHandlerAdvisor { + + private boolean encodingSettingProvided; + private boolean decodingSettingProvided; + + public boolean isDecodingSettingProvided() { + return decodingSettingProvided; + } + + public boolean isEncodingSettingProvided() { + return this.encodingSettingProvided; + } + + @Override + public BindHandler apply(BindHandler bindHandler) { + BindHandler handler = new AbstractBindHandler(bindHandler) { + @Override + public Bindable onStart(ConfigurationPropertyName name, + Bindable target, BindContext context) { + final String configName = name.toString(); + if (configName.contains("use") && configName.contains("native") && + (configName.contains("encoding") || configName.contains("decoding"))) { + BindResult result = context.getBinder().bind(name, target); + if (result.isBound()) { + if (configName.contains("encoding")) { + EncodingDecodingBindAdviceHandler.this.encodingSettingProvided = true; + } + else { + EncodingDecodingBindAdviceHandler.this.decodingSettingProvided = true; + } + return target.withExistingValue(result.get()); + } + } + return bindHandler.onStart(name, target, context); + } + }; + return handler; + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java index b99952dc4..28037ed93 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java @@ -40,10 +40,13 @@ public class GlobalKTableBoundElementFactory extends AbstractBindingTargetFactory { private final BindingServiceProperties bindingServiceProperties; + private final EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler; - GlobalKTableBoundElementFactory(BindingServiceProperties bindingServiceProperties) { + GlobalKTableBoundElementFactory(BindingServiceProperties bindingServiceProperties, + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { super(GlobalKTable.class); this.bindingServiceProperties = bindingServiceProperties; + this.encodingDecodingBindAdviceHandler = encodingDecodingBindAdviceHandler; } @Override @@ -54,6 +57,11 @@ public class GlobalKTableBoundElementFactory consumerProperties = this.bindingServiceProperties.getConsumerProperties(name); consumerProperties.setUseNativeDecoding(true); } + else { + if (!encodingDecodingBindAdviceHandler.isDecodingSettingProvided()) { + consumerProperties.setUseNativeDecoding(true); + } + } // Always set multiplex to true in the kafka streams binder consumerProperties.setMultiplex(true); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index 4f51f7392..d084a1d58 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -22,6 +22,7 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; +import org.apache.kafka.streams.processor.StreamPartitioner; import org.springframework.aop.framework.Advised; import org.springframework.cloud.stream.binder.AbstractBinder; @@ -110,10 +111,11 @@ class KStreamBinder extends protected Binding> doBindProducer(String name, KStream outboundBindTarget, ExtendedProducerProperties properties) { - ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>( - properties.getExtension()); - this.kafkaTopicProvisioner.provisionProducerDestination(name, - extendedProducerProperties); + + ExtendedProducerProperties extendedProducerProperties = + (ExtendedProducerProperties) properties; + + this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties); Serde keySerde = this.keyValueSerdeResolver .getOuboundKeySerde(properties.getExtension(), kafkaStreamsBindingInformationCatalogue.getOutboundKStreamResolvable()); LOG.info("Key Serde used for (outbound) " + name + ": " + keySerde.getClass().getName()); @@ -129,27 +131,36 @@ class KStreamBinder extends LOG.info("Key Serde used for (outbound) " + name + ": " + valueSerde.getClass().getName()); to(properties.isUseNativeEncoding(), name, outboundBindTarget, - (Serde) keySerde, (Serde) valueSerde); + (Serde) keySerde, (Serde) valueSerde, properties.getExtension()); return new DefaultBinding<>(name, null, outboundBindTarget, null); } @SuppressWarnings("unchecked") private void to(boolean isNativeEncoding, String name, - KStream outboundBindTarget, Serde keySerde, - Serde valueSerde) { + KStream outboundBindTarget, Serde keySerde, + Serde valueSerde, KafkaStreamsProducerProperties properties) { + final Produced produced = Produced.with(keySerde, valueSerde); + StreamPartitioner streamPartitioner = null; + if (!StringUtils.isEmpty(properties.getStreamPartitionerBeanName())) { + streamPartitioner = getApplicationContext().getBean(properties.getStreamPartitionerBeanName(), + StreamPartitioner.class); + } + if (streamPartitioner != null) { + produced.withStreamPartitioner(streamPartitioner); + } if (!isNativeEncoding) { LOG.info("Native encoding is disabled for " + name + ". Outbound message conversion done by Spring Cloud Stream."); outboundBindTarget.filter((k, v) -> v == null) - .to(name, Produced.with(keySerde, valueSerde)); + .to(name, produced); this.kafkaStreamsMessageConversionDelegate .serializeOnOutbound(outboundBindTarget) - .to(name, Produced.with(keySerde, valueSerde)); + .to(name, produced); } else { LOG.info("Native encoding is enabled for " + name + ". Outbound serialization done at the broker."); - outboundBindTarget.to(name, Produced.with(keySerde, valueSerde)); + outboundBindTarget.to(name, produced); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java index 526e55dc7..dd6f78b6f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java @@ -43,12 +43,15 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { private final BindingServiceProperties bindingServiceProperties; private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + private final EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler; KStreamBoundElementFactory(BindingServiceProperties bindingServiceProperties, - KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { super(KStream.class); this.bindingServiceProperties = bindingServiceProperties; this.kafkaStreamsBindingInformationCatalogue = KafkaStreamsBindingInformationCatalogue; + this.encodingDecodingBindAdviceHandler = encodingDecodingBindAdviceHandler; } @Override @@ -59,6 +62,11 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { consumerProperties = this.bindingServiceProperties.getConsumerProperties(name); consumerProperties.setUseNativeDecoding(true); } + else { + if (!encodingDecodingBindAdviceHandler.isDecodingSettingProvided()) { + consumerProperties.setUseNativeDecoding(true); + } + } // Always set multiplex to true in the kafka streams binder consumerProperties.setMultiplex(true); return createProxyForKStream(name); @@ -74,6 +82,11 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { producerProperties = this.bindingServiceProperties.getProducerProperties(name); producerProperties.setUseNativeEncoding(true); } + else { + if (!encodingDecodingBindAdviceHandler.isEncodingSettingProvided()) { + producerProperties.setUseNativeEncoding(true); + } + } return createProxyForKStream(name); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java index 533801938..7f45a794e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java @@ -38,10 +38,13 @@ import org.springframework.util.Assert; class KTableBoundElementFactory extends AbstractBindingTargetFactory { private final BindingServiceProperties bindingServiceProperties; + private final EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler; - KTableBoundElementFactory(BindingServiceProperties bindingServiceProperties) { + KTableBoundElementFactory(BindingServiceProperties bindingServiceProperties, + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { super(KTable.class); this.bindingServiceProperties = bindingServiceProperties; + this.encodingDecodingBindAdviceHandler = encodingDecodingBindAdviceHandler; } @Override @@ -52,6 +55,11 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { consumerProperties = this.bindingServiceProperties.getConsumerProperties(name); consumerProperties.setUseNativeDecoding(true); } + else { + if (!encodingDecodingBindAdviceHandler.isDecodingSettingProvided()) { + consumerProperties.setUseNativeDecoding(true); + } + } // Always set multiplex to true in the kafka streams binder consumerProperties.setMultiplex(true); 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 03d2fbb44..ea5b726ee 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 @@ -304,21 +304,22 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean public KStreamBoundElementFactory kStreamBoundElementFactory( BindingServiceProperties bindingServiceProperties, - KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { return new KStreamBoundElementFactory(bindingServiceProperties, - KafkaStreamsBindingInformationCatalogue); + KafkaStreamsBindingInformationCatalogue, encodingDecodingBindAdviceHandler); } @Bean public KTableBoundElementFactory kTableBoundElementFactory( - BindingServiceProperties bindingServiceProperties) { - return new KTableBoundElementFactory(bindingServiceProperties); + BindingServiceProperties bindingServiceProperties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { + return new KTableBoundElementFactory(bindingServiceProperties, encodingDecodingBindAdviceHandler); } @Bean public GlobalKTableBoundElementFactory globalKTableBoundElementFactory( - BindingServiceProperties properties) { - return new GlobalKTableBoundElementFactory(properties); + BindingServiceProperties properties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { + return new GlobalKTableBoundElementFactory(properties, encodingDecodingBindAdviceHandler); } @Bean @@ -375,7 +376,10 @@ public class KafkaStreamsBinderSupportAutoConfiguration { cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties); } - + @Bean + public EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler() { + return new EncodingDecodingBindAdviceHandler(); + } @Configuration @ConditionalOnMissingBean(value = KafkaStreamsBinderMetrics.class, name = "outerContext") diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index 33b2b9e1b..84898ae01 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -65,8 +65,8 @@ final class KafkaStreamsBinderUtils { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, ExtendedConsumerProperties properties) { - ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( - properties.getExtension()); + ExtendedConsumerProperties extendedConsumerProperties = + (ExtendedConsumerProperties) properties; if (binderConfigurationProperties .getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { extendedConsumerProperties.getExtension().setEnableDlq(true); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java index 953f357f0..e5df57aef 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java @@ -36,6 +36,11 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties { */ private String valueSerde; + /** + * {@link org.apache.kafka.streams.processor.StreamPartitioner} to be used on Kafka Streams producer. + */ + private String streamPartitionerBeanName; + public String getKeySerde() { return this.keySerde; } @@ -52,4 +57,11 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties { this.valueSerde = valueSerde; } + public String getStreamPartitionerBeanName() { + return this.streamPartitionerBeanName; + } + + public void setStreamPartitionerBeanName(String streamPartitionerBeanName) { + this.streamPartitionerBeanName = streamPartitionerBeanName; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index d3382cdf3..31b51561a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -31,6 +31,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Serialized; import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.processor.StreamPartitioner; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.ClassRule; @@ -56,7 +57,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { @ClassRule public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts", "counts-1"); + "counts", "counts-1", "counts-2"); private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); @@ -67,9 +68,10 @@ public class KafkaStreamsBinderWordCountFunctionTests { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); consumer = cf.createConsumer(); - embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts", "counts-1"); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts", "counts-1", "counts-2"); } @AfterClass @@ -121,6 +123,45 @@ public class KafkaStreamsBinderWordCountFunctionTests { } } + @Test + public void testKstreamWordCountFunctionWithCustomProducerStreamPartitioner() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.process-in-0.destination=words-2", + "--spring.cloud.stream.bindings.process-out-0.destination=counts-2", + "--spring.cloud.stream.bindings.process-out-0.producer.partitionCount=2", + "--spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.streamPartitionerBeanName" + + "=streamPartitioner", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words-2"); + template.sendDefault("foo"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-2"); + assertThat(cr.value().contains("\"word\":\"foo\",\"count\":1")).isTrue(); + assertThat(cr.partition() == 0) .isTrue(); + template.sendDefault("bar"); + cr = KafkaTestUtils.getSingleRecord(consumer, "counts-2"); + assertThat(cr.value().contains("\"word\":\"bar\",\"count\":1")).isTrue(); + assertThat(cr.partition() == 1) .isTrue(); + } + finally { + pf.destroy(); + } + } + } + private void receiveAndValidate(String in, String out) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); @@ -193,7 +234,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { InteractiveQueryService interactiveQueryService; @Bean - public Function, KStream> process() { + public Function, KStream> process() { return input -> input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) @@ -202,8 +243,13 @@ public class KafkaStreamsBinderWordCountFunctionTests { .windowedBy(TimeWindows.of(5000)) .count(Materialized.as("foo-WordCounts")) .toStream() - .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, + .map((key, value) -> new KeyValue<>(key.key(), new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); } + + @Bean + public StreamPartitioner streamPartitioner() { + return (t, k, v, n) -> k.equals("foo") ? 0 : 1; + } } }