diff --git a/kafka-streams-samples/kafka-streams-inventory-count/README.adoc b/kafka-streams-samples/kafka-streams-inventory-count/README.adoc index 19b4433..246d350 100644 --- a/kafka-streams-samples/kafka-streams-inventory-count/README.adoc +++ b/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. diff --git a/kafka-streams-samples/kafka-streams-inventory-count/pom.xml b/kafka-streams-samples/kafka-streams-inventory-count/pom.xml index 3a6c0ba..af4e909 100644 --- a/kafka-streams-samples/kafka-streams-inventory-count/pom.xml +++ b/kafka-streams-samples/kafka-streams-inventory-count/pom.xml @@ -8,48 +8,60 @@ Kafka Streams inventory count sample - 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 + - 5.5.2 - 2.3.1.RELEASE + Hoxton.BUILD-SNAPSHOT 2.3.1 + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + org.springframework.cloud spring-cloud-stream-binder-kafka-streams - - - org.springframework.kafka - spring-kafka-test - ${spring-kafka-test.version} - test - - 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.apache.kafka kafka-streams-test-utils ${kafka-streams-test-utils.version} test - - org.springframework.cloud - spring-cloud-stream-binder-kafka-streams - + @@ -58,7 +70,19 @@ + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + spring-milestones Spring Milestones @@ -67,16 +91,19 @@ false - - spring-releases - Spring Releases - https://repo.spring.io/release - - false - - + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + spring-milestones Spring Milestones diff --git a/kafka-streams-samples/kafka-streams-inventory-count/src/main/java/kafka/streams/inventory/count/KafkaStreamsInventoryCountApplication.java b/kafka-streams-samples/kafka-streams-inventory-count/src/main/java/kafka/streams/inventory/count/KafkaStreamsInventoryCountApplication.java index 28f177a..0a69021 100644 --- a/kafka-streams-samples/kafka-streams-inventory-count/src/main/java/kafka/streams/inventory/count/KafkaStreamsInventoryCountApplication.java +++ b/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 process(KStream input) { - return input + @Bean + public Function, KStream> process() { + return input -> input .groupByKey(Grouped.with(keySerde, updateEventSerde)) .aggregate(InventoryCountEvent::new, - (key, updateEvent, summaryEvent) -> inventoryCountUpdateEventUpdater.apply(updateEvent, summaryEvent) - // , Materialized.>as(STORE_NAME) - ,Materialized.as(storeSupplier) + (key, updateEvent, summaryEvent) -> inventoryCountUpdateEventUpdater.apply(updateEvent, summaryEvent), + Materialized.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(); - } } diff --git a/kafka-streams-samples/kafka-streams-inventory-count/src/main/resources/application.yml b/kafka-streams-samples/kafka-streams-inventory-count/src/main/resources/application.yml index 3f75732..241ab45 100644 --- a/kafka-streams-samples/kafka-streams-inventory-count/src/main/resources/application.yml +++ b/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 - - diff --git a/kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/TopolologyTestDriverKafkaStreamsInventoryCountTests.java b/kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/TopolologyTestDriverKafkaStreamsInventoryCountTests.java index 04921c0..8f26b44 100644 --- a/kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/TopolologyTestDriverKafkaStreamsInventoryCountTests.java +++ b/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 input = builder.stream(INPUT_TOPIC, Consumed.with(keySerde, updateEventSerde)); KafkaStreamsInventoryAggregator inventoryAggregator = new KafkaStreamsInventoryAggregator(Stores.inMemoryKeyValueStore(STORE_NAME)); - KStream output = inventoryAggregator.process(input); + KStream output = inventoryAggregator.process().apply(input); output.to(OUTPUT_TOPIC); Topology topology = builder.build(); diff --git a/kafka-streams-samples/pom.xml b/kafka-streams-samples/pom.xml index a2acaa9..d6fb709 100644 --- a/kafka-streams-samples/pom.xml +++ b/kafka-streams-samples/pom.xml @@ -20,6 +20,7 @@ kafka-streams-product-tracker kafka-streams-aggregate kafka-streams-to-rabbitmq-message-channel + kafka-streams-inventory-count