Update kafka-streams-to-rabbitmq-message-channel sample

kafka-streams-samples/kafka-streams-to-rabbitmq-message-channel/pom.xml

@@ -11,12 +11,28 @@
 	<description>Demo project for Spring Boot</description>
 
 	<parent>
-		<groupId>io.spring.cloud.stream.sample</groupId>
-		<artifactId>spring-cloud-stream-samples-parent</artifactId>
-		<version>0.0.1-SNAPSHOT</version>
-		<relativePath>../..</relativePath>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.2.0.RELEASE</version>
+		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 
+	<properties>
+		<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
+	</properties>
+
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-dependencies</artifactId>
+				<version>${spring-cloud.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
@@ -24,20 +40,44 @@
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
 		</dependency>
-
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-streams-test-utils</artifactId>
+			<version>${kafka.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
 	</dependencies>
 
+
 	<build>
 		<plugins>
 			<plugin>
@@ -47,4 +87,54 @@
 		</plugins>
 	</build>
 
+	<repositories>
+		<repository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</repository>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+	</repositories>
+	<pluginRepositories>
+		<pluginRepository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-releases</id>
+			<name>Spring Releases</name>
+			<url>https://repo.spring.io/libs-release-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+	</pluginRepositories>
 </project>

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<?, WordCount> process(KStream<Object, String> input) {
+		@Bean
+		public Function<KStream<Object, String>, KStream<?, WordCount>> 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<WordCount, WordCount> sink() {
+			return input -> input;
 		}
 
-		@StreamListener("testInputFromRabbit")
-		public void receive(String data) {
-			System.out.println("Data received..." + data);
+		@Bean
+		public Consumer<String> 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 {

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