Switching the inventory sample to the functional model

kafka-streams-samples/kafka-streams-inventory-count/README.adoc

@@ -18,8 +18,7 @@ compare the following testing strategies:
 * `KafkaStreamsInventoryCountTests` - Uses an Embedded Kafka Broker and manually created Spring application context.
 * `SpringBootKafkaStreamsInventoryCountTests` - Uses an Embedded Kafka Broker and is annotated with `@SpringBootTest`.
 * `TopolologyTestDriverKafkaStreamsInventoryCountTests` - Use the `TopologyTestDriver` and invokes the processer directly.
-
-There is no Embedded Kafka Broker or Spring configuration, so the tests execute very fast.
+There is no Embedded Kafka Broker or Spring configuration in these tests, so the tests execute very fast.
 
 All three implementations run the same set of tests, each processes randomly generated test data.
 

kafka-streams-samples/kafka-streams-inventory-count/pom.xml

@@ -8,48 +8,60 @@
 	<description>Kafka Streams inventory count sample</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>
-		<junit-jupiter.version>5.5.2</junit-jupiter.version>
-		<spring-kafka-test.version>2.3.1.RELEASE</spring-kafka-test.version>
+		<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
 		<kafka-streams-test-utils.version>2.3.1</kafka-streams-test-utils.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>
 			<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
 		</dependency>
-
-		<dependency>
-			<groupId>org.springframework.kafka</groupId>
-			<artifactId>spring-kafka-test</artifactId>
-			<version>${spring-kafka-test.version}</version>
-			<scope>test</scope>
-		</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.apache.kafka</groupId>
 			<artifactId>kafka-streams-test-utils</artifactId>
 			<version>${kafka-streams-test-utils.version}</version>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
-		</dependency>
 	</dependencies>
 
+
 	<build>
 		<plugins>
 			<plugin>
@@ -58,7 +70,19 @@
 			</plugin>
 		</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>
@@ -67,16 +91,19 @@
 				<enabled>false</enabled>
 			</snapshots>
 		</repository>
-		<repository>
-			<id>spring-releases</id>
-			<name>Spring Releases</name>
-			<url>https://repo.spring.io/release</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>

kafka-streams-samples/kafka-streams-inventory-count/src/main/java/kafka/streams/inventory/count/KafkaStreamsInventoryCountApplication.java

@@ -15,25 +15,20 @@
  */
 package kafka.streams.inventory.count;
 
+import java.util.function.Function;
+
 import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.common.utils.Bytes;
 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.state.KeyValueBytesStoreSupplier;
-import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.Stores;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 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.context.annotation.Bean;
 import org.springframework.kafka.support.serializer.JsonSerde;
-import org.springframework.messaging.handler.annotation.SendTo;
 
 
 @SpringBootApplication
@@ -51,8 +46,6 @@ public class KafkaStreamsInventoryCountApplication {
         return Stores.inMemoryKeyValueStore(STORE_NAME);
     }
 
-
-    @EnableBinding(UpdateEventProcessor.class)
     public static class KafkaStreamsInventoryAggregator {
 
         private static final Logger logger = LoggerFactory.getLogger(KafkaStreamsInventoryAggregator.class);
@@ -74,27 +67,16 @@ public class KafkaStreamsInventoryCountApplication {
             this.updateEventSerde = new JsonSerde<>(InventoryUpdateEvent.class);
         }
 
-        @StreamListener("input")
-        @SendTo("output")
-        public KStream<ProductKey, InventoryCountEvent> process(KStream<ProductKey, InventoryUpdateEvent> input) {
-            return input
+        @Bean
+        public Function<KStream<ProductKey, InventoryUpdateEvent>, KStream<ProductKey, InventoryCountEvent>> process() {
+            return input -> input
                     .groupByKey(Grouped.with(keySerde, updateEventSerde))
                     .aggregate(InventoryCountEvent::new,
-                            (key, updateEvent, summaryEvent) -> inventoryCountUpdateEventUpdater.apply(updateEvent, summaryEvent)
-                    //        , Materialized.<ProductKey, InventoryCountEvent, KeyValueStore<Bytes, byte[]>>as(STORE_NAME)
-                              ,Materialized.<ProductKey, InventoryCountEvent>as(storeSupplier)
+                            (key, updateEvent, summaryEvent) -> inventoryCountUpdateEventUpdater.apply(updateEvent, summaryEvent),
+                              Materialized.<ProductKey, InventoryCountEvent>as(storeSupplier)
                                     .withKeySerde(keySerde)
                                     .withValueSerde(countEventSerde))
-
                     .toStream().peek((k, v) -> logger.debug("aggregated count key {} {}", k.getProductCode(), v.getCount()));
         }
     }
-
-    interface UpdateEventProcessor {
-        @Input("input")
-        KStream<?, ?> input();
-
-        @Output("output")
-        KStream<?, ?> output();
-    }
 }

kafka-streams-samples/kafka-streams-inventory-count/src/main/resources/application.yml

@@ -1,14 +1,6 @@
-spring.application.name: kafka-streams-aggregate-sample
+spring.application.name: kafka-streams-inventory-processor
 
-spring.cloud.stream.bindings.input:
+spring.cloud.stream.bindings.process-in-0:
   destination: inventory-update-events
-  group: inventory-processor
-spring.cloud.stream.bindings.output:
+spring.cloud.stream.bindings.process-out-0:
   destination: inventory-count-events
-
-spring.cloud.stream.kafka.streams.binder:
-  configuration:
-    spring.json.trusted.packages: kafka.streams.inventory.count
-    default.key.serde: org.springframework.kafka.support.serializer.JsonSerde
-
-

kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/TopolologyTestDriverKafkaStreamsInventoryCountTests.java

@@ -82,7 +82,7 @@ public class TopolologyTestDriverKafkaStreamsInventoryCountTests extends Abstrac
         KStream<ProductKey, InventoryUpdateEvent> input = builder.stream(INPUT_TOPIC, Consumed.with(keySerde, updateEventSerde));
         KafkaStreamsInventoryAggregator inventoryAggregator = new KafkaStreamsInventoryAggregator(Stores.inMemoryKeyValueStore(STORE_NAME));
 
-        KStream<ProductKey, InventoryCountEvent> output = inventoryAggregator.process(input);
+        KStream<ProductKey, InventoryCountEvent> output = inventoryAggregator.process().apply(input);
         output.to(OUTPUT_TOPIC);
 
         Topology topology = builder.build();

kafka-streams-samples/pom.xml

@@ -20,6 +20,7 @@
         <module>kafka-streams-product-tracker</module>
         <module>kafka-streams-aggregate</module>
         <module>kafka-streams-to-rabbitmq-message-channel</module>
+        <module>kafka-streams-inventory-count</module>
     </modules>
 
 </project>