Add inventory count sample

Clean up

Change name in pom.xml

kafka-streams-samples/kafka-streams-inventory-count/.gitignore

@@ -0,0 +1,24 @@
+target/
+!.mvn/wrapper/maven-wrapper.jar
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+nbproject/private/
+build/
+nbbuild/
+dist/
+nbdist/
+.nb-gradle/

kafka-streams-samples/kafka-streams-inventory-count/.mvn/wrapper/maven-wrapper.properties

@@ -0,0 +1 @@
+distributionUrl=https://repo1.maven.org/maven2/org/apache/maven/apache-maven/3.5.0/apache-maven-3.5.0-bin.zip

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

@@ -0,0 +1,34 @@
+== What is this app?
+
+This is an example of a Spring Cloud Stream processor using Kafka Streams aggregation.
+
+The application illustrates an inventory tracking use case. InventoryUpdateEvents are input and keyed by ProductKey.
+
+Each event contains the key, a delta value, an action:
+
+   * `INC` - add the delta to the existing count
+   * `DEC` - subtract the delta from the existing count
+   * `REP` - replace the count with the delta value
+
+The output topic contains a running total for each product key.
+
+This sample illustrates the use of Custom key and Value types with Json serialization. We also
+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.
+
+All three implementations run the same set of tests, each processes randomly generated test data.
+
+=== Running the tests:
+
+```bash
+$./mvnw clean test
+```
+
+
+
+

kafka-streams-samples/kafka-streams-inventory-count/mvnw

@@ -0,0 +1,225 @@
+#!/bin/sh
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+# ----------------------------------------------------------------------------
+
+# ----------------------------------------------------------------------------
+# Maven2 Start Up Batch script
+#
+# Required ENV vars:
+# ------------------
+#   JAVA_HOME - location of a JDK home dir
+#
+# Optional ENV vars
+# -----------------
+#   M2_HOME - location of maven2's installed home dir
+#   MAVEN_OPTS - parameters passed to the Java VM when running Maven
+#     e.g. to debug Maven itself, use
+#       set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+#   MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+# ----------------------------------------------------------------------------
+
+if [ -z "$MAVEN_SKIP_RC" ] ; then
+
+  if [ -f /etc/mavenrc ] ; then
+    . /etc/mavenrc
+  fi
+
+  if [ -f "$HOME/.mavenrc" ] ; then
+    . "$HOME/.mavenrc"
+  fi
+
+fi
+
+# OS specific support.  $var _must_ be set to either true or false.
+cygwin=false;
+darwin=false;
+mingw=false
+case "`uname`" in
+  CYGWIN*) cygwin=true ;;
+  MINGW*) mingw=true;;
+  Darwin*) darwin=true
+    # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
+    # See https://developer.apple.com/library/mac/qa/qa1170/_index.html
+    if [ -z "$JAVA_HOME" ]; then
+      if [ -x "/usr/libexec/java_home" ]; then
+        export JAVA_HOME="`/usr/libexec/java_home`"
+      else
+        export JAVA_HOME="/Library/Java/Home"
+      fi
+    fi
+    ;;
+esac
+
+if [ -z "$JAVA_HOME" ] ; then
+  if [ -r /etc/gentoo-release ] ; then
+    JAVA_HOME=`java-config --jre-home`
+  fi
+fi
+
+if [ -z "$M2_HOME" ] ; then
+  ## resolve links - $0 may be a link to maven's home
+  PRG="$0"
+
+  # need this for relative symlinks
+  while [ -h "$PRG" ] ; do
+    ls=`ls -ld "$PRG"`
+    link=`expr "$ls" : '.*-> \(.*\)$'`
+    if expr "$link" : '/.*' > /dev/null; then
+      PRG="$link"
+    else
+      PRG="`dirname "$PRG"`/$link"
+    fi
+  done
+
+  saveddir=`pwd`
+
+  M2_HOME=`dirname "$PRG"`/..
+
+  # make it fully qualified
+  M2_HOME=`cd "$M2_HOME" && pwd`
+
+  cd "$saveddir"
+  # echo Using m2 at $M2_HOME
+fi
+
+# For Cygwin, ensure paths are in UNIX format before anything is touched
+if $cygwin ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --unix "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
+fi
+
+# For Migwn, ensure paths are in UNIX format before anything is touched
+if $mingw ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME="`(cd "$M2_HOME"; pwd)`"
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
+  # TODO classpath?
+fi
+
+if [ -z "$JAVA_HOME" ]; then
+  javaExecutable="`which javac`"
+  if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
+    # readlink(1) is not available as standard on Solaris 10.
+    readLink=`which readlink`
+    if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
+      if $darwin ; then
+        javaHome="`dirname \"$javaExecutable\"`"
+        javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
+      else
+        javaExecutable="`readlink -f \"$javaExecutable\"`"
+      fi
+      javaHome="`dirname \"$javaExecutable\"`"
+      javaHome=`expr "$javaHome" : '\(.*\)/bin'`
+      JAVA_HOME="$javaHome"
+      export JAVA_HOME
+    fi
+  fi
+fi
+
+if [ -z "$JAVACMD" ] ; then
+  if [ -n "$JAVA_HOME"  ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+      # IBM's JDK on AIX uses strange locations for the executables
+      JAVACMD="$JAVA_HOME/jre/sh/java"
+    else
+      JAVACMD="$JAVA_HOME/bin/java"
+    fi
+  else
+    JAVACMD="`which java`"
+  fi
+fi
+
+if [ ! -x "$JAVACMD" ] ; then
+  echo "Error: JAVA_HOME is not defined correctly." >&2
+  echo "  We cannot execute $JAVACMD" >&2
+  exit 1
+fi
+
+if [ -z "$JAVA_HOME" ] ; then
+  echo "Warning: JAVA_HOME environment variable is not set."
+fi
+
+CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
+
+# traverses directory structure from process work directory to filesystem root
+# first directory with .mvn subdirectory is considered project base directory
+find_maven_basedir() {
+
+  if [ -z "$1" ]
+  then
+    echo "Path not specified to find_maven_basedir"
+    return 1
+  fi
+
+  basedir="$1"
+  wdir="$1"
+  while [ "$wdir" != '/' ] ; do
+    if [ -d "$wdir"/.mvn ] ; then
+      basedir=$wdir
+      break
+    fi
+    # workaround for JBEAP-8937 (on Solaris 10/Sparc)
+    if [ -d "${wdir}" ]; then
+      wdir=`cd "$wdir/.."; pwd`
+    fi
+    # end of workaround
+  done
+  echo "${basedir}"
+}
+
+# concatenates all lines of a file
+concat_lines() {
+  if [ -f "$1" ]; then
+    echo "$(tr -s '\n' ' ' < "$1")"
+  fi
+}
+
+BASE_DIR=`find_maven_basedir "$(pwd)"`
+if [ -z "$BASE_DIR" ]; then
+  exit 1;
+fi
+
+export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
+echo $MAVEN_PROJECTBASEDIR
+MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
+
+# For Cygwin, switch paths to Windows format before running java
+if $cygwin; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --path --windows "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
+  [ -n "$MAVEN_PROJECTBASEDIR" ] &&
+    MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
+fi
+
+WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+exec "$JAVACMD" \
+  $MAVEN_OPTS \
+  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
+  "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
+  ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"

kafka-streams-samples/kafka-streams-inventory-count/mvnw.cmd

@@ -0,0 +1,143 @@
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    http://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Maven2 Start Up Batch script
+@REM
+@REM Required ENV vars:
+@REM JAVA_HOME - location of a JDK home dir
+@REM
+@REM Optional ENV vars
+@REM M2_HOME - location of maven2's installed home dir
+@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
+@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a key stroke before ending
+@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
+@REM     e.g. to debug Maven itself, use
+@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+@REM ----------------------------------------------------------------------------
+
+@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
+@echo off
+@REM enable echoing my setting MAVEN_BATCH_ECHO to 'on'
+@if "%MAVEN_BATCH_ECHO%" == "on"  echo %MAVEN_BATCH_ECHO%
+
+@REM set %HOME% to equivalent of $HOME
+if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
+
+@REM Execute a user defined script before this one
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
+@REM check for pre script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat"
+if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd"
+:skipRcPre
+
+@setlocal
+
+set ERROR_CODE=0
+
+@REM To isolate internal variables from possible post scripts, we use another setlocal
+@setlocal
+
+@REM ==== START VALIDATION ====
+if not "%JAVA_HOME%" == "" goto OkJHome
+
+echo.
+echo Error: JAVA_HOME not found in your environment. >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+:OkJHome
+if exist "%JAVA_HOME%\bin\java.exe" goto init
+
+echo.
+echo Error: JAVA_HOME is set to an invalid directory. >&2
+echo JAVA_HOME = "%JAVA_HOME%" >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+@REM ==== END VALIDATION ====
+
+:init
+
+@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
+@REM Fallback to current working directory if not found.
+
+set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
+IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
+
+set EXEC_DIR=%CD%
+set WDIR=%EXEC_DIR%
+:findBaseDir
+IF EXIST "%WDIR%"\.mvn goto baseDirFound
+cd ..
+IF "%WDIR%"=="%CD%" goto baseDirNotFound
+set WDIR=%CD%
+goto findBaseDir
+
+:baseDirFound
+set MAVEN_PROJECTBASEDIR=%WDIR%
+cd "%EXEC_DIR%"
+goto endDetectBaseDir
+
+:baseDirNotFound
+set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
+cd "%EXEC_DIR%"
+
+:endDetectBaseDir
+
+IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
+
+@setlocal EnableExtensions EnableDelayedExpansion
+for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
+@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
+
+:endReadAdditionalConfig
+
+SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
+
+set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
+set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
+if ERRORLEVEL 1 goto error
+goto end
+
+:error
+set ERROR_CODE=1
+
+:end
+@endlocal & set ERROR_CODE=%ERROR_CODE%
+
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost
+@REM check for post script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat"
+if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd"
+:skipRcPost
+
+@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
+if "%MAVEN_BATCH_PAUSE%" == "on" pause
+
+if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE%
+
+exit /B %ERROR_CODE%

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

@@ -0,0 +1,98 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>kafka-streams-inventory-count</artifactId>
+	<packaging>jar</packaging>
+	<name>kafka-streams-inventory-count</name>
+	<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>
+	</parent>
+
+	<properties>
+		<junit-jupiter.version>5.5.2</junit-jupiter.version>
+		<spring-kafka-test.version>2.3.1.RELEASE</spring-kafka-test.version>
+		<kafka-streams-test-utils.version>2.3.1</kafka-streams-test-utils.version>
+	</properties>
+
+	<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.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>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+	<repositories>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<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-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-inventory-count/src/main/java/kafka/streams/inventory/count/InventoryCountEvent.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.Objects;
+
+/**
+ * @author David Turanski
+ */
+public class InventoryCountEvent {
+
+	private int count;
+
+	private ProductKey key;
+
+
+	public InventoryCountEvent(){
+	};
+
+
+	public InventoryCountEvent(ProductKey key, int count) {
+		this.count = count;
+		this.key = key;
+	}
+
+	public int getCount() {
+		return count;
+	}
+
+	public void setCount(int count) {
+		this.count = count;
+	}
+
+	public ProductKey getKey() {
+		return key;
+	}
+
+	public void setKey(ProductKey key) {
+		this.key = key;
+	}
+
+
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) return true;
+		if (o == null || getClass() != o.getClass()) return false;
+		InventoryCountEvent that = (InventoryCountEvent) o;
+		return count == that.count &&
+				Objects.equals(key, that.key);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(count, key);
+	}
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.function.BiFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * Function to apply a {@link InventoryUpdateEvent} to an existing {@link InventoryCountEvent}.
+ *
+ * This is used by the stream processor and also by the test harness to compute the expected count, given a sequence of generated events.
+ *
+ * @author David Turanski
+ */
+public class InventoryCountUpdateEventUpdater implements BiFunction<InventoryUpdateEvent, InventoryCountEvent, InventoryCountEvent> {
+    private final static Logger logger = LoggerFactory.getLogger(InventoryCountUpdateEventUpdater.class);
+
+    @Override
+    public InventoryCountEvent apply(InventoryUpdateEvent inventoryUpdateEvent, InventoryCountEvent inventoryCountEvent) {
+        int delta = inventoryUpdateEvent.getDelta();
+        logger.trace("Applying update {} {} {} to inventoryCountEvent. Current count is {}",
+                inventoryUpdateEvent.getKey().getProductCode(), inventoryUpdateEvent.getAction(), inventoryUpdateEvent.getDelta(), inventoryCountEvent.getCount());
+        inventoryCountEvent.setKey(inventoryUpdateEvent.getKey());
+        switch (inventoryUpdateEvent.getAction()) {
+            case DEC:
+                inventoryCountEvent.setCount(inventoryCountEvent.getCount() - delta);
+                break;
+            case INC:
+                inventoryCountEvent.setCount(inventoryCountEvent.getCount() + delta);
+                break;
+            case REP:
+                inventoryCountEvent.setCount(delta);
+                break;
+            default:
+                return null;
+        }
+        logger.trace("Applied update {} {} {} to inventoryCountEvent. Current count is {}",
+                inventoryUpdateEvent.getKey().getProductCode(), inventoryUpdateEvent.getAction(), inventoryUpdateEvent.getDelta(), inventoryCountEvent.getCount());
+        return inventoryCountEvent;
+    }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+/**
+ * @author David Turanski
+ */
+
+public class InventoryUpdateEvent {
+
+	public InventoryUpdateEvent() {
+		return;
+	}
+
+	private int delta;
+
+	private ProductKey key;
+
+	private Action action;
+
+	public enum Action{
+		INC, DEC, REP;
+	}
+
+	public int getDelta() {
+		return delta;
+	}
+
+	public void setDelta(int delta) {
+		this.delta = delta;
+	}
+
+	public ProductKey getKey() {
+		return key;
+	}
+
+	public void setKey(ProductKey key) {
+		this.key = key;
+	}
+
+	public Action getAction() {
+		return action;
+	}
+
+	public void setAction(Action action) {
+		this.action = action;
+	}
+}

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+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
+public class KafkaStreamsInventoryCountApplication {
+
+
+    final static String STORE_NAME = "inventory-counts";
+
+    public static void main(String[] args) {
+        SpringApplication.run(KafkaStreamsInventoryAggregator.class, args);
+    }
+
+    @Bean
+    public KeyValueBytesStoreSupplier storeSupplier() {
+        return Stores.inMemoryKeyValueStore(STORE_NAME);
+    }
+
+
+    @EnableBinding(UpdateEventProcessor.class)
+    public static class KafkaStreamsInventoryAggregator {
+
+        private static final Logger logger = LoggerFactory.getLogger(KafkaStreamsInventoryAggregator.class);
+
+        private final KeyValueBytesStoreSupplier storeSupplier;
+
+        private final InventoryCountUpdateEventUpdater inventoryCountUpdateEventUpdater = new InventoryCountUpdateEventUpdater();
+
+        private final  Serde<InventoryCountEvent> countEventSerde;
+
+        private final Serde<InventoryUpdateEvent> updateEventSerde;
+
+        private final  Serde<ProductKey> keySerde;
+
+        public KafkaStreamsInventoryAggregator(KeyValueBytesStoreSupplier storeSupplier) {
+            this.storeSupplier = storeSupplier;
+            this.keySerde = new JsonSerde<>(ProductKey.class);
+            this.countEventSerde = new JsonSerde<>(InventoryCountEvent.class);
+            this.updateEventSerde = new JsonSerde<>(InventoryUpdateEvent.class);
+        }
+
+        @StreamListener("input")
+        @SendTo("output")
+        public KStream<ProductKey, InventoryCountEvent> process(KStream<ProductKey, InventoryUpdateEvent> input) {
+            return 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)
+                                    .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/java/kafka/streams/inventory/count/ProductKey.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.Objects;
+
+public class ProductKey {
+    private String productCode;
+
+    public ProductKey() {
+    }
+
+
+    public ProductKey(String productCode) {
+        this.productCode = productCode;
+    }
+
+    public String getProductCode() {
+        return productCode;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ProductKey key = (ProductKey) o;
+        return Objects.equals(productCode, key.productCode);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(productCode);
+    }
+}

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

@@ -0,0 +1,14 @@
+spring.application.name: kafka-streams-aggregate-sample
+
+spring.cloud.stream.bindings.input:
+  destination: inventory-update-events
+  group: inventory-processor
+spring.cloud.stream.bindings.output:
+  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/main/resources/logback.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{ISO8601} %5p [%.-10t] %c{2}:%L - %m%n</pattern>
+        </encoder>
+    </appender>
+    <root level="WARN">
+        <appender-ref ref="stdout"/>
+    </root>
+    <logger name="org.apache.kafka.streams.processor.internals" level="WARN"/>
+    <logger name="kafka.streams.inventory.count" level="DEBUG"/>
+    <logger name="org.springframework.kafka.config" level="DEBUG"/>
+</configuration>

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

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import kafka.streams.inventory.count.generator.AbstractInventoryUpdateEventGenerator;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.RepeatedTest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Base class for running various implementations of aggregation tests.
+ * Each test is repeated multiple times to ensure that state is consistent through multiple invocations.
+ *
+ * @author David Turanski
+ */
+public abstract class AbstractInventoryCountTests {
+    private static final int REPETITION_COUNT = 3;
+
+    private static AbstractInventoryUpdateEventGenerator eventGenerator;
+
+    protected Consumer<ProductKey, InventoryCountEvent> consumer;
+
+    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+
+    /**
+     *
+     * @param eventGenerator an {@link AbstractInventoryUpdateEventGenerator} implementation.
+     */
+    protected static void setEventGenerator(AbstractInventoryUpdateEventGenerator eventGenerator) {
+        AbstractInventoryCountTests.eventGenerator = eventGenerator;
+    }
+
+    @RepeatedTest(REPETITION_COUNT)
+    public void processMessagesForSingleKey() {
+
+        Map<ProductKey, InventoryCountEvent> expectedCounts = eventGenerator.generateRandomEvents(1, 3);
+
+        Map<ProductKey, InventoryCountEvent> actualEvents = consumeActualInventoryCountEvents(3);
+
+        assertThat(actualEvents).hasSize(1);
+
+        expectedCounts.forEach((key, value) ->
+                assertThat(actualEvents.get(key).getCount()).isEqualTo(value.getCount()));
+    }
+
+    @RepeatedTest(REPETITION_COUNT)
+    public void processAggregatedEventsForSingleKey() {
+        Map<ProductKey, InventoryCountEvent> expectedCounts;
+        expectedCounts = eventGenerator.generateRandomEvents(1, 5);
+
+        Map<ProductKey, InventoryCountEvent> originalCount = consumeActualInventoryCountEvents(5);
+
+        expectedCounts.forEach((key, value) ->
+                assertThat(originalCount.get(key).getCount()).isEqualTo(value.getCount()));
+
+        expectedCounts = eventGenerator.generateRandomEvents(1, 5);
+
+        Map<ProductKey, InventoryCountEvent> actualCount = consumeActualInventoryCountEvents(5);
+
+        expectedCounts.forEach((key, value) ->
+                assertThat(actualCount.get(key).getCount()).isEqualTo(value.getCount()));
+    }
+
+    @RepeatedTest(REPETITION_COUNT)
+    public void processAggregatedEventsForMultipleKeys() {
+        Map<ProductKey, InventoryCountEvent> initialCounts = eventGenerator.generateRandomEvents(10, 5);
+
+        Map<ProductKey, InventoryCountEvent> expectedEvents;
+        expectedEvents = consumeActualInventoryCountEvents(50);
+
+        expectedEvents.forEach((key, value) ->
+                assertThat(initialCounts.get(key).getCount()).isEqualTo(value.getCount()));
+
+        Map<ProductKey, InventoryCountEvent> updatedCounts = eventGenerator.generateRandomEvents(10, 5);
+
+        expectedEvents = consumeActualInventoryCountEvents(50);
+
+        boolean atLeastOneUpdatedCountIsDifferent = false;
+
+        for (ProductKey key : updatedCounts.keySet()) {
+            assertThat(expectedEvents.get(key).getCount()).isEqualTo(updatedCounts.get(key).getCount());
+            atLeastOneUpdatedCountIsDifferent = atLeastOneUpdatedCountIsDifferent || !initialCounts.get(key).equals(updatedCounts.get(key));
+        }
+
+        //Verify that the expected counts changed from the first round of events.
+        assertThat(atLeastOneUpdatedCountIsDifferent).isTrue();
+    }
+
+    /**
+     * Reset the state by sending 0 count values to the aggregator.
+     * These events are also consumed so that subsequent tests do not have to deal with additional events.
+     */
+    @AfterEach
+    void tearDown() {
+        Map<ProductKey, InventoryCountEvent> events = eventGenerator.reset();
+        consumeActualInventoryCountEvents(events.size());
+        if (consumer != null) {
+            consumer.close();
+        }
+    }
+
+    /**
+     * Consume the actual events from the output topic.
+     * This implementation uses a {@link Consumer}, assuming a (an embedded) Kafka Broker but may be overridden.
+     * @param expectedCount the expected number of messages is known. This avoids a timeout delay if all is well.
+     *
+     * @return the consumed data.
+     */
+    protected Map<ProductKey, InventoryCountEvent> consumeActualInventoryCountEvents(int expectedCount) {
+        Map<ProductKey, InventoryCountEvent> inventoryCountEvents = new LinkedHashMap<>();
+        int receivedCount = 0;
+        while (receivedCount < expectedCount) {
+            ConsumerRecords<ProductKey, InventoryCountEvent> records = KafkaTestUtils.getRecords(consumer, 1000);
+            if (records.isEmpty()) {
+                logger.error("No more records received. Expected {} received {}.", expectedCount, receivedCount);
+                break;
+            }
+            receivedCount += records.count();
+            for (Iterator<ConsumerRecord<ProductKey, InventoryCountEvent>> it = records.iterator(); it.hasNext(); ) {
+                ConsumerRecord<ProductKey, InventoryCountEvent> consumerRecord = it.next();
+                logger.debug("consumed " + consumerRecord.key().getProductCode() + " = " + consumerRecord.value().getCount());
+                inventoryCountEvents.put(consumerRecord.key(), consumerRecord.value());
+            }
+        }
+        return inventoryCountEvents;
+    }
+}

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import kafka.streams.inventory.count.generator.KafkaTemplateInventoryUpdateEventGenerator;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * A test implementation that uses {@link SpringApplicationBuilder} directly, instead of '@SpringBootTest'.
+ * The advantage is that the {@link EmbeddedKafkaBroker} can be provided by Junit 5 to an `@BeforeAll` method, and the
+ * Spring context is configured accordingly.
+ *
+ * Note, the base class closes the consumer after each test.
+ *
+ * @author David Turanski
+ */
+@EmbeddedKafka(topics = KafkaStreamsInventoryCountTests.INPUT_TOPIC)
+public class KafkaStreamsInventoryCountTests extends AbstractInventoryCountTests{
+
+    static final String INPUT_TOPIC = "inventory-update-events";
+    static final String OUTPUT_TOPIC = "inventory-count-events";
+    private static final String GROUP_NAME = "inventory-count-test";
+
+    private static ConfigurableApplicationContext context;
+    private static DefaultKafkaConsumerFactory<ProductKey, InventoryCountEvent> cf;
+
+    @BeforeAll
+    public static void init(EmbeddedKafkaBroker embeddedKafka) {
+        Map<String, Object> props = new HashMap<>();
+
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        setEventGenerator(new KafkaTemplateInventoryUpdateEventGenerator(props, INPUT_TOPIC));
+
+        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(GROUP_NAME, "true", embeddedKafka);
+        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "test");
+        consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
+        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, KafkaStreamsInventoryCountTests.class.getPackage().getName());
+        consumerProps.put(JsonDeserializer.KEY_DEFAULT_TYPE, ProductKey.class);
+        consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, InventoryCountEvent.class);
+        consumerProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, "false");
+        cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+
+
+        /*
+         * Disabling caching makes the test run faster, and more consistent behavior with the TopologyTestDriver.
+         * More messages are produced on the output topic.
+         */
+        context = new SpringApplicationBuilder(KafkaStreamsInventoryCountApplication.class)
+                .properties(
+                        "spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
+                        "spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+                        "spring.cloud.stream.kafka.streams.binder.configuration.cache.max.bytes.buffering=0")
+                .run();
+    }
+
+    @AfterAll
+    public static void shutdown() {
+        context.close();
+    }
+
+    @BeforeEach
+    public void setUp() {
+        consumer = cf.createConsumer(GROUP_NAME);
+        consumer.subscribe(Collections.singleton(OUTPUT_TOPIC));
+    }
+}

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import kafka.streams.inventory.count.generator.AbstractInventoryUpdateEventGenerator;
+import kafka.streams.inventory.count.generator.KafkaTemplateInventoryUpdateEventGenerator;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.BeforeEach;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+
+/**
+ * A test implementation annotated with '@SpringBootTest'.
+ *
+ * The Spring context is implicitly created and is auto configured to use {@link EmbeddedKafkaBroker}, configured with the required `bootStrapServers` property.
+ * Here the EmbeddedKafkaBroker must be autowired as a instance variable, so not available for a static '@BeforeAll' method.
+ * Consequently, the {@link DefaultKafkaConsumerFactory} which depends on the broker is configured in `@BeforeEach`.
+ *
+ * Note, the base class closes the consumer after each test.
+ *
+ * @author David Turanski
+ */
+@EmbeddedKafka(
+        bootstrapServersProperty = "spring.kafka.bootstrap-servers",
+        topics = {
+                SpringBootKafkaStreamsInventoryCountTests.INPUT_TOPIC
+        })
+/*
+ * Disabling caching makes the test run faster, and more consistent behavior with the TopologyTestDriver.
+ * More messages are produced on the output topic.
+ */
+@SpringBootTest(
+        properties = {
+                "spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+                "spring.cloud.stream.kafka.streams.binder.configuration.cache.max.bytes.buffering=0"
+        })
+public class SpringBootKafkaStreamsInventoryCountTests extends AbstractInventoryCountTests {
+
+    static final String INPUT_TOPIC = "inventory-update-events";
+    static final String OUTPUT_TOPIC = "inventory-count-events";
+    private static final String GROUP_NAME = "inventory-count-test";
+
+    private DefaultKafkaConsumerFactory<ProductKey, InventoryCountEvent> cf;
+
+    @Autowired
+    private EmbeddedKafkaBroker broker;
+
+    @BeforeEach
+    void setUp() {
+
+        Map<String, Object> props = new HashMap<>();
+
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        AbstractInventoryUpdateEventGenerator eventGenerator = new
+                KafkaTemplateInventoryUpdateEventGenerator(props, INPUT_TOPIC);
+        setEventGenerator(eventGenerator);
+
+        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(GROUP_NAME, "true", broker);
+        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "test");
+        consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
+        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, KafkaStreamsInventoryCountTests.class.getPackage().getName());
+        consumerProps.put(JsonDeserializer.KEY_DEFAULT_TYPE, ProductKey.class);
+        consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, InventoryCountEvent.class);
+        consumerProps.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, "false");
+        cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+
+        consumer = cf.createConsumer(GROUP_NAME);
+        consumer.subscribe(Collections.singleton(OUTPUT_TOPIC));
+    }
+}

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

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+import kafka.streams.inventory.count.KafkaStreamsInventoryCountApplication.KafkaStreamsInventoryAggregator;
+import kafka.streams.inventory.count.generator.TopologyTestDriverUpdateEventGenerator;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerde;
+
+import static kafka.streams.inventory.count.KafkaStreamsInventoryCountApplication.STORE_NAME;
+
+/**
+ * A test implementation that uses {@link TopologyTestDriver}. There is no Spring configuration or embedded Kafka broker
+ * here so the execution time is very fast. The process method is invoked directly and everything is run in a single thread.
+ *
+ *
+ *
+ * @author David Turanski
+ */
+public class TopolologyTestDriverKafkaStreamsInventoryCountTests extends AbstractInventoryCountTests {
+
+    static final String INPUT_TOPIC = "inventory-update-events";
+    static final String OUTPUT_TOPIC = "inventory-count-events";
+
+    private Serde<InventoryCountEvent> countEventSerde = new JsonSerde<>(InventoryCountEvent.class);
+    private Serde<InventoryUpdateEvent> updateEventSerde = new JsonSerde<>(InventoryUpdateEvent.class);
+    private Serde<ProductKey> keySerde = new JsonSerde<>(ProductKey.class);
+
+    private TopologyTestDriver testDriver;
+
+    static Properties getStreamsConfiguration() {
+        final Properties streamsConfiguration = new Properties();
+        // Need to be set even these do not matter with TopologyTestDriver
+        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "TopologyTestDriver");
+        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "ignored");
+        return streamsConfiguration;
+    }
+
+    private void configureDeserializer(Deserializer<?> deserializer, Class<?> keyDefaultType, Class<?> valueDefaultType, boolean isKey) {
+        Map<String, Object> deserializerConfig = new HashMap<>();
+        deserializerConfig.put(JsonDeserializer.KEY_DEFAULT_TYPE, keyDefaultType);
+        deserializerConfig.put(JsonDeserializer.VALUE_DEFAULT_TYPE, valueDefaultType);
+        deserializer.configure(deserializerConfig, isKey);
+    }
+
+    @BeforeEach
+     void setup() {
+        configureDeserializer(countEventSerde.deserializer(), ProductKey.class, InventoryCountEvent.class, false);
+        configureDeserializer(keySerde.deserializer() ,ProductKey.class, null, true);
+
+        final StreamsBuilder builder = new StreamsBuilder();
+
+        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);
+        output.to(OUTPUT_TOPIC);
+
+        Topology topology = builder.build();
+        testDriver = new TopologyTestDriver(topology, getStreamsConfiguration());
+
+        logger.debug(topology.describe().toString());
+
+
+        setEventGenerator(new TopologyTestDriverUpdateEventGenerator(testDriver, INPUT_TOPIC, keySerde.serializer(),
+                updateEventSerde.serializer()));
+    }
+
+    @AfterEach
+    void tearDown() {
+        super.tearDown();
+        try {
+            testDriver.close();
+        } catch (final RuntimeException e) {
+            // https://issues.apache.org/jira/browse/KAFKA-6647 causes exception when executed in Windows, ignoring it
+            // Logged stacktrace cannot be avoided
+            System.out.println("Ignoring exception, test failing in Windows due this exception:" + e.getLocalizedMessage());
+        }
+    }
+
+    @Override
+    protected Map<ProductKey, InventoryCountEvent> consumeActualInventoryCountEvents(int expectedCount) {
+        Map<ProductKey, InventoryCountEvent> inventoryCountEvents = new LinkedHashMap<>();
+        int receivedCount = 0;
+        while (receivedCount < expectedCount) {
+            ProducerRecord<ProductKey, InventoryCountEvent> record
+                    = testDriver.readOutput(OUTPUT_TOPIC, keySerde.deserializer(), countEventSerde.deserializer());
+            if (record == null) {
+                break;
+            }
+            receivedCount++;
+            logger.debug("consumed " + record.key().getProductCode() + " = " + record.value().getCount());
+            inventoryCountEvents.put(record.key(), record.value());
+        }
+        return inventoryCountEvents;
+    }
+
+}

kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/generator/AbstractInventoryUpdateEventGenerator.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count.generator;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Random;
+import kafka.streams.inventory.count.InventoryCountEvent;
+import kafka.streams.inventory.count.InventoryCountUpdateEventUpdater;
+import kafka.streams.inventory.count.InventoryUpdateEvent;
+import kafka.streams.inventory.count.ProductKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static kafka.streams.inventory.count.InventoryUpdateEvent.Action.DEC;
+import static kafka.streams.inventory.count.InventoryUpdateEvent.Action.INC;
+import static kafka.streams.inventory.count.InventoryUpdateEvent.Action.REP;
+
+/**
+ * Base class to generate random {@link InventoryUpdateEvent}s which are aggregated by the stream processor.
+ * Subclasses implement 'doSendEvent(key,value)'.
+ *
+ * @author David Turanski
+ */
+public abstract class AbstractInventoryUpdateEventGenerator {
+
+    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private final Map<ProductKey, InventoryCountEvent> accumulatedInventoryCounts = new LinkedHashMap<>();
+
+    public Map<ProductKey, InventoryCountEvent> generateRandomEvents(int numberKeys, int eventsPerKey) {
+        InventoryUpdateEvent.Action[] actions = {INC, DEC, REP};
+        return doGenerateEvents(numberKeys, eventsPerKey, actions);
+    }
+
+    /**
+     * Resets the Kafka stream materialized state and the internal state by sending a 0 count for each existing key.
+     * @return the state prior to invoking this method.
+     */
+    public Map<ProductKey, InventoryCountEvent> reset() {
+        Map<ProductKey, InventoryCountEvent> current
+                = Collections.unmodifiableMap(new LinkedHashMap(accumulatedInventoryCounts));
+
+        accumulatedInventoryCounts.keySet().forEach(key -> {
+            InventoryUpdateEvent inventoryUpdateEvent = new InventoryUpdateEvent();
+            inventoryUpdateEvent.setKey(key);
+            inventoryUpdateEvent.setAction(REP);
+            inventoryUpdateEvent.setDelta(0);
+            sendEvent(key, inventoryUpdateEvent);
+        });
+        accumulatedInventoryCounts.clear();
+        return current;
+    }
+
+    /**
+     * @param numberKeys   number of keys to generate events for.
+     * @param eventsPerKey number of events per key.
+     * @param actions      the list of update actions to include.
+     * @return expected calculated counts. Accumulates values since last reset to simulate what the aggregator does.
+     */
+    private Map<ProductKey, InventoryCountEvent> doGenerateEvents(int numberKeys, int eventsPerKey, InventoryUpdateEvent.Action[] actions) {
+        Random random = new Random();
+
+        InventoryCountUpdateEventUpdater summaryEventUpdater = new InventoryCountUpdateEventUpdater();
+
+        for (int j = 0; j < numberKeys; j++) {
+            ProductKey key = new ProductKey("key" + j);
+            InventoryCountEvent inventoryCountEvent = new InventoryCountEvent(key,
+                    accumulatedInventoryCounts.containsKey(key) ? accumulatedInventoryCounts.get(key).getCount() : 0);
+            for (int i = 0; i < eventsPerKey; i++) {
+                InventoryUpdateEvent inventoryUpdateEvent = new InventoryUpdateEvent();
+                inventoryUpdateEvent.setKey(key);
+
+                inventoryUpdateEvent.setDelta(random.nextInt(10) + 1);
+                inventoryUpdateEvent.setAction(actions[random.nextInt(actions.length)]);
+
+                inventoryCountEvent = summaryEventUpdater.apply(inventoryUpdateEvent, inventoryCountEvent);
+
+                sendEvent(inventoryUpdateEvent.getKey(),inventoryUpdateEvent);
+
+            }
+            accumulatedInventoryCounts.put(key, inventoryCountEvent);
+        }
+
+        return Collections.unmodifiableMap(new LinkedHashMap<>(accumulatedInventoryCounts));
+
+    }
+
+    protected void sendEvent(ProductKey key, InventoryUpdateEvent value) {
+        logger.debug("Sending inventoryUpdateEvent: key {} delta {} action {}",
+                key.getProductCode(), value.getDelta(), value.getAction());
+       doSendEvent(key, value);
+    }
+
+    protected abstract void doSendEvent(ProductKey key, InventoryUpdateEvent value);
+}

kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/generator/KafkaTemplateInventoryUpdateEventGenerator.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count.generator;
+
+import java.util.Map;
+import kafka.streams.inventory.count.InventoryUpdateEvent;
+import kafka.streams.inventory.count.ProductKey;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Test data generator using {@link InventoryUpdateEvent}s using {@link KafkaTemplate} to send events.
+ * Used for testing with {@link org.springframework.kafka.test.EmbeddedKafkaBroker}.
+ *
+ * @author David Turanski
+ */
+public class KafkaTemplateInventoryUpdateEventGenerator extends AbstractInventoryUpdateEventGenerator {
+
+    private final KafkaTemplate<ProductKey, InventoryUpdateEvent> kafkaTemplate;
+
+    public KafkaTemplateInventoryUpdateEventGenerator(Map<String, Object> producerProperties, String destination) {
+        DefaultKafkaProducerFactory<ProductKey, InventoryUpdateEvent> pf = new DefaultKafkaProducerFactory(producerProperties);
+        kafkaTemplate = new KafkaTemplate<>(pf, true);
+        kafkaTemplate.setDefaultTopic(destination);
+    }
+
+    protected void doSendEvent(ProductKey key, InventoryUpdateEvent value) {
+        kafkaTemplate.sendDefault(key, value);
+    }
+}

kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count/generator/TopologyTestDriverUpdateEventGenerator.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 kafka.streams.inventory.count.generator;
+
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import kafka.streams.inventory.count.InventoryUpdateEvent;
+import kafka.streams.inventory.count.ProductKey;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.test.ConsumerRecordFactory;
+
+/**
+ * Test data generator using {@link InventoryUpdateEvent}s using {@link TopologyTestDriver} to send events.
+ * Used for testing with {@link TopologyTestDriver}.
+ *
+ * @author David Turanski
+ */
+public class TopologyTestDriverUpdateEventGenerator extends AbstractInventoryUpdateEventGenerator {
+
+    private final TopologyTestDriver topologyTestDriver;
+    private final ConsumerRecordFactory<ProductKey, InventoryUpdateEvent> recordFactory;
+
+    public TopologyTestDriverUpdateEventGenerator(TopologyTestDriver topologyTestDriver,
+                                                  String inputTopic,
+                                                  Serializer<ProductKey> keySerializer,
+                                                  Serializer<InventoryUpdateEvent> valueSerializer) {
+        this.topologyTestDriver = topologyTestDriver;
+        this.recordFactory = new ConsumerRecordFactory<>(
+                 inputTopic, keySerializer, valueSerializer);
+    }
+
+    @Override
+    protected void doSendEvent(ProductKey key, InventoryUpdateEvent value) {
+        ConsumerRecord<byte[], byte[]> record = recordFactory.create(key, value, LocalDateTime.now().toEpochSecond(ZoneOffset.UTC));
+        topologyTestDriver.pipeInput(record);
+    }
+}