diff --git a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/pom.xml b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/pom.xml index 7ad0b9f..13c5977 100644 --- a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/pom.xml +++ b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/pom.xml @@ -11,12 +11,28 @@ Demo project for Spring Boot - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.RELEASE + + + Hoxton.BUILD-SNAPSHOT + + + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + org.springframework.cloud @@ -24,20 +40,44 @@ org.springframework.cloud - spring-cloud-stream-binder-kafka + spring-cloud-stream-binder-rabbit org.springframework.cloud - spring-cloud-stream-binder-rabbit + spring-cloud-stream-binder-kafka - org.springframework.boot spring-boot-starter-test test + + org.springframework.kafka + spring-kafka-test + test + + + org.apache.kafka + kafka-streams-test-utils + ${kafka.version} + test + + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + + @@ -47,4 +87,54 @@ + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release-local + + false + + + diff --git a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/java/kafka/streams/message/channel/KafkaStreamsWordCountApplication.java b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/java/kafka/streams/message/channel/KafkaStreamsWordCountApplication.java index d667888..8540833 100644 --- a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/java/kafka/streams/message/channel/KafkaStreamsWordCountApplication.java +++ b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/java/kafka/streams/message/channel/KafkaStreamsWordCountApplication.java @@ -16,25 +16,21 @@ package kafka.streams.message.channel; -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.apache.kafka.streams.kstream.TimeWindows; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.handler.annotation.SendTo; - +import java.time.Duration; import java.util.Arrays; import java.util.Date; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Grouped; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; @SpringBootApplication public class KafkaStreamsWordCountApplication { @@ -43,60 +39,30 @@ public class KafkaStreamsWordCountApplication { SpringApplication.run(KafkaStreamsWordCountApplication.class, args); } - @EnableBinding(MultipleProcessor.class) public static class WordCountProcessorApplication { - @StreamListener("kstreamIn") - @SendTo("kstreamOut") - public KStream process(KStream input) { + @Bean + public Function, KStream> process() { - return input + return input -> input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) - .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) - .windowedBy(TimeWindows.of(20_000)) + .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) + .windowedBy(TimeWindows.of(Duration.ofSeconds(20))) .count(Materialized.as("WordCounts-1")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); } - @StreamListener("fromKafka") - @SendTo("toRabbit") - public WordCount sink(WordCount input) { - return input; + @Bean + public Function sink() { + return input -> input; } - @StreamListener("testInputFromRabbit") - public void receive(String data) { - System.out.println("Data received..." + data); + @Bean + public Consumer receive() { + return data -> System.out.println("Data received..." + data); } - - } - - interface MultipleProcessor { - - String KSTREAM_IN = "kstreamIn"; - String KSTREAM_OUT = "kstreamOut"; - - String FROM_KAFKA = "fromKafka"; - String TO_RABBIT = "toRabbit"; - - String TEST_INPUT_FROM_RABBIT = "testInputFromRabbit"; - - @Input(KSTREAM_IN) - KStream kstreamIn(); - - @Output(KSTREAM_OUT) - KStream kstreamOut(); - - @Input(FROM_KAFKA) - SubscribableChannel fromKafka(); - - @Output(TO_RABBIT) - MessageChannel toRabbit(); - - @Input(TEST_INPUT_FROM_RABBIT) - SubscribableChannel testInputFromRabbit(); } static class WordCount { diff --git a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/resources/application.yml b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/resources/application.yml index f2541d4..d7f1594 100644 --- a/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/resources/application.yml +++ b/kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/src/main/resources/application.yml @@ -1,23 +1,23 @@ -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 - default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde - bindings.kstreamIn.consumer.application-id: kafka-streams-to-rabbitmq -spring.cloud.stream.bindings.kstreamIn: - destination: words -spring.cloud.stream.bindings.kstreamOut: - destination: counts -spring.cloud.stream.bindings.fromKafka: - destination: counts - binder: kafka -spring.cloud.stream.bindings.toRabbit: - destination: countsInRabbit - binder: rabbit +spring.cloud.stream: + function.definition: process;sink;receive + kafka.streams: + binder.configuration: + commit.interval.ms: 1000 + default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde + default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde + bindings: + process-in-0: + destination: words + process-out-0: + destination: counts + sink-in-0: + destination: counts + binder: kafka + sink-out-0: + destination: countsInRabbit + binder: rabbit + receive-in-0: + destination: countsInRabbit + binder: rabbit -spring.cloud.stream.bindings.testInputFromRabbit: - destination: countsInRabbit - binder: rabbit - -spring.cloud.stream.kafka.streams.binder: - brokers: localhost +spring.application.name: kafka-streams-to-rabbitmq \ No newline at end of file