From 894597309b1537025066a77e7d9012a4b0fad54b Mon Sep 17 00:00:00 2001 From: Sarath Shyam Date: Sat, 14 Apr 2018 17:30:24 +0530 Subject: [PATCH] Fix kaka-streams binder to consume messages from multiple input topics #361 Modified KafkaStreamsStreamListenerSetupMethodOrchestrator#getkStream so that the KStream object is built from list of topic names --- ...StreamListenerSetupMethodOrchestrator.java | 5 +- ...aStreamsBinderMultipleInputTopicsTest.java | 195 ++++++++++++++++++ 2 files changed, 199 insertions(+), 1 deletion(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMultipleInputTopicsTest.java 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 7caa21a12..9cdd0458c 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.lang.reflect.Method; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.Map; @@ -341,7 +342,9 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene LOG.info("state store " + storeBuilder.name() + " added to topology"); } } - KStream stream = streamsBuilder.stream(bindingServiceProperties.getBindingDestination(inboundName), + String[] bindingTargets = StringUtils + .commaDelimitedListToStringArray(bindingServiceProperties.getBindingDestination(inboundName)); + KStream stream = streamsBuilder.stream(Arrays.asList(bindingTargets), Consumed.with(keySerde, valueSerde)); final boolean nativeDecoding = bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding(); if (nativeDecoding){ diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMultipleInputTopicsTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMultipleInputTopicsTest.java new file mode 100644 index 000000000..848c7b689 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMultipleInputTopicsTest.java @@ -0,0 +1,195 @@ +/* + * Copyright 2017 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 + * + * http://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 java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Serialized; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +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.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Sarath Shyam + * + * This test case demonstrates a kafk-streams topology which consumes messages from + * multiple kafka topics(destinations). + * See {@link KafkaStreamsBinderMultipleInputTopicsTest#testKstreamWordCountWithStringInputAndPojoOuput} where + * the input topic names are specified as comma-separated String values for + * the property spring.cloud.stream.bindings.input.destination. + * + * + */ +public class KafkaStreamsBinderMultipleInputTopicsTest { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=words1,words2", + "--spring.cloud.stream.bindings.output.destination=counts", + "--spring.cloud.stream.bindings.output.contentType=application/json", + "--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.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", + "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + try { + receiveAndValidate(context); + } finally { + context.close(); + } + } + + private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words1"); + template.sendDefault("foobar1"); + template.setDefaultTopic("words2"); + template.sendDefault("foobar2"); + + //Sleep a bit so that both the messages are processed before reading from the output topic. + //Else assertions might fail arbitrarily. + Thread.sleep(5000); + + ConsumerRecords received = KafkaTestUtils.getRecords(consumer); + + List wordCounts = new ArrayList<>(2); + + received.records("counts").forEach((consumerRecord) -> { + wordCounts.add((consumerRecord.value())); + }); + System.out.println(wordCounts); + assertThat(wordCounts.contains("{\"word\":\"foobar1\",\"count\":1}")).isTrue(); + assertThat(wordCounts.contains("{\"word\":\"foobar2\",\"count\":1}")).isTrue(); + + } + + @EnableBinding(KafkaStreamsProcessor.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) + static class WordCountProcessorApplication { + + + @StreamListener + @SendTo("output") + 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 + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .map((key, value) -> new KeyValue<>(value, value)) + .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) + .count(Materialized.as("WordCounts")) + .toStream() + .map((key, value) -> new KeyValue<>(null, new WordCount(key, value))); + } + + } + + static class WordCount { + + private String word; + + private long count; + + WordCount(String word, long count) { + this.word = word; + this.count = count; + } + + public String getWord() { + return word; + } + + public void setWord(String word) { + this.word = word; + } + + public long getCount() { + return count; + } + + public void setCount(long count) { + this.count = count; + } + + } + +}