From 44793b83bef0cdadf0c67d7fdb4abcc30ac75610 Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Mon, 28 Sep 2015 13:34:51 -0500
Subject: [PATCH 1/6] KAFKA-2593 Key value stores can use custom serializers
 and deserializers

Add support for the key value stores to use specified serializers and deserializers (aka, "serdes"). Prior to this change, the stores were limited to only the default serdes specified in the topology's configuration and exposed to the processors via the ProcessorContext.

Now, using InMemoryKeyValueStore and RocksDBKeyValueStore are similar: both are parameterized on the key and value types, and both have similar multiple static factory methods. The static factory methods either take explicit key and value serdes, take key and value class types so the serdes can be inferred (only for the built-in serdes for string, integer, long, and byte array types), or use the default serdes on the ProcessorContext.
---
 .../kafka/streams/examples/ProcessorJob.java  |   2 +-
 .../streams/state/InMemoryKeyValueStore.java  | 129 ++++++++++-
 .../streams/state/MeteredKeyValueStore.java   |  17 +-
 .../streams/state/RocksDBKeyValueStore.java   | 203 ++++++++++++++----
 .../apache/kafka/streams/state/Serdes.java    | 164 ++++++++++++++
 .../internals/ProcessorTopologyTest.java      |   2 +-
 6 files changed, 458 insertions(+), 59 deletions(-)
 create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/Serdes.java

diff --git a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
index 92e6284e73cfe..cdfc29b0e93be 100644
--- a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
+++ b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
@@ -48,7 +48,7 @@ public Processor<String, String> get() {
                 public void init(ProcessorContext context) {
                     this.context = context;
                     this.context.schedule(1000);
-                    this.kvStore = new InMemoryKeyValueStore<>("local-state", context);
+                    this.kvStore = InMemoryKeyValueStore.create("local-state", context, String.class, Integer.class);
                 }
 
                 @Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
index 59a8496de2966..01cc53f68000d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
@@ -17,9 +17,11 @@
 
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.processor.ProcessorContext;
 
 import java.util.Iterator;
 import java.util.List;
@@ -35,26 +37,133 @@
  */
 public class InMemoryKeyValueStore<K, V> extends MeteredKeyValueStore<K, V> {
 
-    public InMemoryKeyValueStore(String name, ProcessorContext context) {
-        this(name, context, new SystemTime());
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic and the system time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @return the key-value store
+     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
+     *             {@code Integer.class}, {@code Long.class}, or
+     *             {@code byte[].class}
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass) {
+        return create(name, context, keyClass, valueClass, new SystemTime());
+    }
+
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic and the given time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
+     *             {@code Integer.class}, {@code Long.class}, or
+     *             {@code byte[].class}
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
+                                                            Time time) {
+        return new InMemoryKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), time);
+    }
+
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default
+     * serializers and deserializers, and the system time provider.
+     * <p>
+     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
+     * used as parameters for this key value store. This is not checked in this method, and any mismatch will result in
+     * class cast exceptions during usage.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @return the key-value store
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context) {
+        return create(name, context, new SystemTime());
+    }
+
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default
+     * serializers and deserializers, and the given time provider.
+     * <p>
+     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
+     * used as parameters for this key value store.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Time time) {
+        Serdes<K, V> serdes = new Serdes<>(name, context);
+        return new InMemoryKeyValueStore<>(name, context, serdes, time);
+    }
+
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and
+     * the system time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keySerializer the serializer for keys; may not be null
+     * @param keyDeserializer the deserializer for keys; may not be null
+     * @param valueSerializer the serializer for values; may not be null
+     * @param valueDeserializer the deserializer for values; may not be null
+     * @return the key-value store
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context,
+                                                            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+                                                            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) {
+        return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime());
+    }
+
+    /**
+     * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and
+     * the given time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keySerializer the serializer for keys; may not be null
+     * @param keyDeserializer the deserializer for keys; may not be null
+     * @param valueSerializer the serializer for values; may not be null
+     * @param valueDeserializer the deserializer for values; may not be null
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context,
+                                                            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+                                                            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer,
+                                                            Time time) {
+        Serdes<K, V> serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
+        return new InMemoryKeyValueStore<>(name, context, serdes, time);
     }
 
-    public InMemoryKeyValueStore(String name, ProcessorContext context, Time time) {
-        super(name, new MemoryStore<K, V>(name, context), context, "in-memory-state", time);
+    protected InMemoryKeyValueStore(String name, ProcessorContext context, Serdes<K, V> serdes, Time time) {
+        super(name, new MemoryStore<K, V>(name), context, serdes, "in-memory-state", time != null ? time : new SystemTime());
     }
 
     private static class MemoryStore<K, V> implements KeyValueStore<K, V> {
 
         private final String name;
         private final NavigableMap<K, V> map;
-        private final ProcessorContext context;
 
-        @SuppressWarnings("unchecked")
-        public MemoryStore(String name, ProcessorContext context) {
+        public MemoryStore(String name) {
             super();
             this.name = name;
             this.map = new TreeMap<>();
-            this.context = context;
         }
 
         @Override
@@ -137,4 +246,4 @@ public void close() {
 
         }
     }
-}
+}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
index 68333d5c6bd7a..f45594e3d3fbd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
@@ -35,6 +35,7 @@
 public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
 
     protected final KeyValueStore<K, V> inner;
+    protected final Serdes<K, V> serialization;
 
     private final Time time;
     private final Sensor putTime;
@@ -54,8 +55,10 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
     private final ProcessorContext context;
 
     // always wrap the logged store with the metered store
-    public MeteredKeyValueStore(final String name, final KeyValueStore<K, V> inner, ProcessorContext context, String metricGrp, Time time) {
+    public MeteredKeyValueStore(final String name, final KeyValueStore<K, V> inner, ProcessorContext context,
+                                Serdes<K, V> serialization, String metricGrp, Time time) {
         this.inner = inner;
+        this.serialization = serialization;
 
         this.time = time;
         this.metrics = context.metrics();
@@ -79,8 +82,8 @@ public MeteredKeyValueStore(final String name, final KeyValueStore<K, V> inner,
         // register and possibly restore the state from the logs
         long startNs = time.nanoseconds();
         try {
-            final Deserializer<K> keyDeserializer = (Deserializer<K>) context.keyDeserializer();
-            final Deserializer<V> valDeserializer = (Deserializer<V>) context.valueDeserializer();
+            final Deserializer<K> keyDeserializer = serialization.keyDeserializer();
+            final Deserializer<V> valDeserializer = serialization.valueDeserializer();
 
             context.register(this, new StateRestoreCallback() {
                 @Override
@@ -189,10 +192,10 @@ public void flush() {
 
     private void logChange() {
         RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
-        Serializer<K> keySerializer = (Serializer<K>) context.keySerializer();
-        Serializer<V> valueSerializer = (Serializer<V>) context.valueSerializer();
-
         if (collector != null) {
+            Serializer<K> keySerializer = serialization.keySerializer();
+            Serializer<V> valueSerializer = serialization.valueSerializer();
+
             for (K k : this.dirty) {
                 V v = this.inner.get(k);
                 collector.send(new ProducerRecord<>(this.topic, this.partition, k, v), keySerializer, valueSerializer);
@@ -239,4 +242,4 @@ public void close() {
 
     }
 
-}
+}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
index 373bba07b0442..fbd7002c6db50 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
@@ -17,11 +17,12 @@
 
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.SystemTime;
-
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.processor.ProcessorContext;
 import org.rocksdb.BlockBasedTableConfig;
 import org.rocksdb.CompactionStyle;
 import org.rocksdb.CompressionType;
@@ -37,17 +38,135 @@
 import java.util.List;
 import java.util.NoSuchElementException;
 
-public class RocksDBKeyValueStore extends MeteredKeyValueStore<byte[], byte[]> {
+/**
+ * A {@link KeyValueStore} that stores all entries in a local RocksDB database.
+ *
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ */
+public class RocksDBKeyValueStore<K, V> extends MeteredKeyValueStore<K, V> {
+
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the system time
+     * provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @return the key-value store
+     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
+     *             {@code Integer.class}, {@code Long.class}, or
+     *             {@code byte[].class}
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass) {
+        return create(name, context, keyClass, valueClass, new SystemTime());
+    }
+
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the given time
+     * provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
+     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+     *            {@code byte[].class})
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
+     *             {@code Integer.class}, {@code Long.class}, or
+     *             {@code byte[].class}
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
+                                                           Time time) {
+        return new RocksDBKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), time);
+    }
+
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
+     * {@link ProcessorContext}'s default serializers and deserializers, and the system time provider.
+     * <p>
+     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
+     * used as parameters for this key value store. This is not checked in this method, and any mismatch will result in
+     * class cast exceptions during usage.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @return the key-value store
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context) {
+        return create(name, context, new SystemTime());
+    }
 
-    public RocksDBKeyValueStore(String name, ProcessorContext context) {
-        this(name, context, new SystemTime());
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
+     * {@link ProcessorContext}'s default serializers and deserializers, and the given time provider.
+     * <p>
+     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
+     * used as parameters for this key value store.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Time time) {
+        Serdes<K, V> serdes = new Serdes<>(name, context);
+        return new RocksDBKeyValueStore<>(name, context, serdes, time);
     }
 
-    public RocksDBKeyValueStore(String name, ProcessorContext context, Time time) {
-        super(name, new RocksDBStore(name, context), context, "rocksdb-state", time);
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
+     * supplied serializers and deserializers, and the system time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keySerializer the serializer for keys; may not be null
+     * @param keyDeserializer the deserializer for keys; may not be null
+     * @param valueSerializer the serializer for values; may not be null
+     * @param valueDeserializer the deserializer for values; may not be null
+     * @return the key-value store
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context,
+                                                           Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+                                                           Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) {
+        return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime());
     }
 
-    private static class RocksDBStore implements KeyValueStore<byte[], byte[]> {
+    /**
+     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
+     * supplied serializers and deserializers, and the given time provider.
+     * 
+     * @param name the name of the store, used in the name of the topic to which entries are recorded
+     * @param context the processing context
+     * @param keySerializer the serializer for keys; may not be null
+     * @param keyDeserializer the deserializer for keys; may not be null
+     * @param valueSerializer the serializer for values; may not be null
+     * @param valueDeserializer the deserializer for values; may not be null
+     * @param time the time provider; may be null if the system time should be used
+     * @return the key-value store
+     */
+    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context,
+                                                           Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+                                                           Serializer<V> valueSerializer, Deserializer<V> valueDeserializer,
+                                                           Time time) {
+        Serdes<K, V> serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
+        return new RocksDBKeyValueStore<>(name, context, serdes, time);
+    }
+
+    protected RocksDBKeyValueStore(String name, ProcessorContext context, Serdes<K, V> serdes, Time time) {
+        super(name, new RocksDBStore<K, V>(name, context, serdes), context, serdes, "rocksdb-state", time != null ? time : new SystemTime());
+    }
+
+    private static class RocksDBStore<K, V> implements KeyValueStore<K, V> {
 
         private static final int TTL_NOT_USED = -1;
 
@@ -61,6 +180,8 @@ private static class RocksDBStore implements KeyValueStore<byte[], byte[]> {
         private static final int MAX_WRITE_BUFFERS = 3;
         private static final String DB_FILE_DIR = "rocksdb";
 
+        private final Serdes<K, V> serdes;
+
         private final String topic;
         private final int partition;
         private final ProcessorContext context;
@@ -74,11 +195,11 @@ private static class RocksDBStore implements KeyValueStore<byte[], byte[]> {
 
         private RocksDB db;
 
-        @SuppressWarnings("unchecked")
-        public RocksDBStore(String name, ProcessorContext context) {
+        public RocksDBStore(String name, ProcessorContext context, Serdes<K, V> serdes) {
             this.topic = name;
             this.partition = context.id();
             this.context = context;
+            this.serdes = serdes;
 
             // initialize the rocksdb options
             BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
@@ -132,9 +253,9 @@ public boolean persistent() {
         }
 
         @Override
-        public byte[] get(byte[] key) {
+        public V get(K key) {
             try {
-                return this.db.get(key);
+                return serdes.valueFrom(this.db.get(serdes.rawKey(key)));
             } catch (RocksDBException e) {
                 // TODO: this needs to be handled more accurately
                 throw new KafkaException("Error while executing get " + key.toString() + " from store " + this.topic, e);
@@ -142,12 +263,12 @@ public byte[] get(byte[] key) {
         }
 
         @Override
-        public void put(byte[] key, byte[] value) {
+        public void put(K key, V value) {
             try {
                 if (value == null) {
-                    db.remove(wOptions, key);
+                    db.remove(wOptions, serdes.rawKey(key));
                 } else {
-                    db.put(wOptions, key, value);
+                    db.put(wOptions, serdes.rawKey(key), serdes.rawValue(value));
                 }
             } catch (RocksDBException e) {
                 // TODO: this needs to be handled more accurately
@@ -156,28 +277,28 @@ public void put(byte[] key, byte[] value) {
         }
 
         @Override
-        public void putAll(List<Entry<byte[], byte[]>> entries) {
-            for (Entry<byte[], byte[]> entry : entries)
+        public void putAll(List<Entry<K, V>> entries) {
+            for (Entry<K, V> entry : entries)
                 put(entry.key(), entry.value());
         }
-
+        
         @Override
-        public byte[] delete(byte[] key) {
-            byte[] value = get(key);
+        public V delete(K key) {
+            V value = get(key);
             put(key, null);
             return value;
         }
 
         @Override
-        public KeyValueIterator<byte[], byte[]> range(byte[] from, byte[] to) {
-            return new RocksDBRangeIterator(db.newIterator(), from, to);
+        public KeyValueIterator<K, V> range(K from, K to) {
+            return new RocksDBRangeIterator<K, V>(db.newIterator(), serdes, from, to);
         }
 
         @Override
-        public KeyValueIterator<byte[], byte[]> all() {
+        public KeyValueIterator<K, V> all() {
             RocksIterator innerIter = db.newIterator();
             innerIter.seekToFirst();
-            return new RocksDbIterator(innerIter);
+            return new RocksDbIterator<K, V>(innerIter, serdes);
         }
 
         @Override
@@ -196,19 +317,21 @@ public void close() {
             db.close();
         }
 
-        private static class RocksDbIterator implements KeyValueIterator<byte[], byte[]> {
+        private static class RocksDbIterator<K, V> implements KeyValueIterator<K, V> {
             private final RocksIterator iter;
+            private final Serdes<K, V> serdes;
 
-            public RocksDbIterator(RocksIterator iter) {
+            public RocksDbIterator(RocksIterator iter, Serdes<K, V> serdes) {
                 this.iter = iter;
+                this.serdes = serdes;
             }
 
-            protected byte[] peekKey() {
-                return this.getEntry().key();
+            protected byte[] peekRawKey() {
+                return iter.key();
             }
 
-            protected Entry<byte[], byte[]> getEntry() {
-                return new Entry<>(iter.key(), iter.value());
+            protected Entry<K, V> getEntry() {
+                return new Entry<>(serdes.keyFrom(iter.key()), serdes.valueFrom(iter.value()));
             }
 
             @Override
@@ -217,13 +340,12 @@ public boolean hasNext() {
             }
 
             @Override
-            public Entry<byte[], byte[]> next() {
+            public Entry<K, V> next() {
                 if (!hasNext())
                     throw new NoSuchElementException();
 
-                Entry<byte[], byte[]> entry = this.getEntry();
+                Entry<K, V> entry = this.getEntry();
                 iter.next();
-
                 return entry;
             }
 
@@ -253,24 +375,25 @@ public int compare(byte[] left, byte[] right) {
             }
         }
 
-        private static class RocksDBRangeIterator extends RocksDbIterator {
+        private static class RocksDBRangeIterator<K, V> extends RocksDbIterator<K, V> {
             // RocksDB's JNI interface does not expose getters/setters that allow the
             // comparator to be pluggable, and the default is lexicographic, so it's
             // safe to just force lexicographic comparator here for now.
             private final Comparator<byte[]> comparator = new LexicographicComparator();
             byte[] to;
 
-            public RocksDBRangeIterator(RocksIterator iter, byte[] from, byte[] to) {
-                super(iter);
-                iter.seek(from);
-                this.to = to;
+            public RocksDBRangeIterator(RocksIterator iter, Serdes<K, V> serdes,
+                    K from, K to) {
+                super(iter, serdes);
+                iter.seek(serdes.rawKey(from));
+                this.to = serdes.rawKey(to);
             }
 
             @Override
             public boolean hasNext() {
-                return super.hasNext() && comparator.compare(super.peekKey(), this.to) < 0;
+                return super.hasNext() && comparator.compare(super.peekRawKey(), this.to) < 0;
             }
         }
 
     }
-}
+}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
new file mode 100644
index 0000000000000..4fa21a78f397a
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
@@ -0,0 +1,164 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+final class Serdes<K, V> {
+
+    public static <K, V> Serdes<K, V> withBuiltinTypes(String topic, Class<K> keyClass, Class<V> valueClass) {
+        Serializer<K> keySerializer = serializer(keyClass);
+        Deserializer<K> keyDeserializer = deserializer(keyClass);
+        Serializer<V> valueSerializer = serializer(valueClass);
+        Deserializer<V> valueDeserializer = deserializer(valueClass);
+        return new Serdes<>(topic, keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> Serializer<T> serializer(Class<T> type) {
+        if (String.class.isAssignableFrom(type)) return (Serializer<T>) new StringSerializer();
+        if (Integer.class.isAssignableFrom(type)) return (Serializer<T>) new IntegerSerializer();
+        if (Long.class.isAssignableFrom(type)) return (Serializer<T>) new LongSerializer();
+        if (byte[].class.isAssignableFrom(type)) return (Serializer<T>) new ByteArraySerializer();
+        throw new IllegalArgumentException("Unknown class for built-in serializer");
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> Deserializer<T> deserializer(Class<T> type) {
+        if (String.class.isAssignableFrom(type)) return (Deserializer<T>) new StringDeserializer();
+        if (Integer.class.isAssignableFrom(type)) return (Deserializer<T>) new IntegerDeserializer();
+        if (Long.class.isAssignableFrom(type)) return (Deserializer<T>) new LongDeserializer();
+        if (byte[].class.isAssignableFrom(type)) return (Deserializer<T>) new ByteArrayDeserializer();
+        throw new IllegalArgumentException("Unknown class for built-in serializer");
+    }
+
+    private final String topic;
+    private final Serializer<K> keySerializer;
+    private final Serializer<V> valueSerializer;
+    private final Deserializer<K> keyDeserializer;
+    private final Deserializer<V> valueDeserializer;
+
+    /**
+     * Create a context for serialization using the specified serializers and deserializers.
+     * 
+     * @param topic the name of the topic
+     * @param keySerializer the serializer for keys; may not be null
+     * @param keyDeserializer the deserializer for keys; may not be null
+     * @param valueSerializer the serializer for values; may not be null
+     * @param valueDeserializer the deserializer for values; may not be null
+     */
+    public Serdes(String topic,
+            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) {
+        this.topic = topic;
+        this.keySerializer = keySerializer;
+        this.keyDeserializer = keyDeserializer;
+        this.valueSerializer = valueSerializer;
+        this.valueDeserializer = valueDeserializer;
+    }
+
+    /**
+     * Create a context for serialization using the specified serializers and deserializers, or if any of them are null the
+     * corresponding {@link ProcessorContext}'s default serializer or deserializer, which
+     * <em>must</em> match the key and value types used as parameters for this object.
+     * 
+     * @param topic the name of the topic
+     * @param keySerializer the serializer for keys; may be null if the {@link ProcessorContext#keySerializer() default
+     *            key serializer} should be used
+     * @param keyDeserializer the deserializer for keys; may be null if the {@link ProcessorContext#keyDeserializer() default
+     *            key deserializer} should be used
+     * @param valueSerializer the serializer for values; may be null if the {@link ProcessorContext#valueSerializer() default
+     *            value serializer} should be used
+     * @param valueDeserializer the deserializer for values; may be null if the {@link ProcessorContext#valueDeserializer()
+     *            default value deserializer} should be used
+     * @param context the processing context
+     */
+    @SuppressWarnings("unchecked")
+    public Serdes(String topic,
+            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
+            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer,
+            ProcessorContext context) {
+        this.topic = topic;
+        this.keySerializer = keySerializer != null ? keySerializer : (Serializer<K>) context.keySerializer();
+        this.keyDeserializer = keyDeserializer != null ? keyDeserializer : (Deserializer<K>) context.keyDeserializer();
+        this.valueSerializer = valueSerializer != null ? valueSerializer : (Serializer<V>) context.valueSerializer();
+        this.valueDeserializer = valueDeserializer != null ? valueDeserializer : (Deserializer<V>) context.valueDeserializer();
+    }
+
+    /**
+     * Create a context for serialization using the {@link ProcessorContext}'s default serializers and deserializers, which
+     * <em>must</em> match the key and value types used as parameters for this object.
+     * 
+     * @param topic the name of the topic
+     * @param context the processing context
+     */
+    @SuppressWarnings("unchecked")
+    public Serdes(String topic,
+            ProcessorContext context) {
+        this.topic = topic;
+        this.keySerializer = (Serializer<K>) context.keySerializer();
+        this.keyDeserializer = (Deserializer<K>) context.keyDeserializer();
+        this.valueSerializer = (Serializer<V>) context.valueSerializer();
+        this.valueDeserializer = (Deserializer<V>) context.valueDeserializer();
+    }
+
+    public Deserializer<K> keyDeserializer() {
+        return keyDeserializer;
+    }
+
+    public Serializer<K> keySerializer() {
+        return keySerializer;
+    }
+
+    public Deserializer<V> valueDeserializer() {
+        return valueDeserializer;
+    }
+
+    public Serializer<V> valueSerializer() {
+        return valueSerializer;
+    }
+
+    public String topic() {
+        return topic;
+    }
+
+    public K keyFrom(byte[] rawKey) {
+        return keyDeserializer.deserialize(topic, rawKey);
+    }
+
+    public V valueFrom(byte[] rawValue) {
+        return valueDeserializer.deserialize(topic, rawValue);
+    }
+
+    public byte[] rawKey(K key) {
+        return keySerializer.serialize(topic, key);
+    }
+
+    public byte[] rawValue(V value) {
+        return valueSerializer.serialize(topic, value);
+    }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index 50a23ecbbe750..c44fe13e902b9 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -289,7 +289,7 @@ public StatefulProcessor(String storeName) {
         @Override
         public void init(ProcessorContext context) {
             super.init(context);
-            store = new InMemoryKeyValueStore<>(storeName, context);
+            store = InMemoryKeyValueStore.create(storeName, context, String.class, String.class);
         }
 
         @Override

From 2a99b584d4ab156a97c51165fe12511b646d12b4 Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Wed, 30 Sep 2015 16:56:51 -0500
Subject: [PATCH 2/6] KAFKA-2593: MeteredKeyValueStore and others no longer
 cast to ProcessorContextImpl

The cast was used to access ProcessorContextImpl.recordCollector(), and the cast prevent using MeteredKeyValueStore subclasses, SinkNode, and SlidingWindow objects in tests where something other than ProcessorContextImpl was used. This change introduces a RecordCollector.Supplier interface to define this `recordCollector()` method, and changed ProcessorContextImpl and MockProcessorContext to both implement this interface. Now, MeteredKeyValueStore, SinkNode, and SlidingWindow can cast to the interface to access the record collector, making it possible to use them inside unit tests.
---
 .../kstream/SlidingWindowSupplier.java        |  3 +-
 .../internals/ProcessorContextImpl.java       |  3 +-
 .../processor/internals/RecordCollector.java  | 11 ++++
 .../streams/processor/internals/SinkNode.java |  2 +-
 .../streams/state/MeteredKeyValueStore.java   |  3 +-
 .../kafka/test/MockProcessorContext.java      | 54 +++++++++++++++----
 6 files changed, 60 insertions(+), 16 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SlidingWindowSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SlidingWindowSupplier.java
index 0110c875a1419..bf6b4dcd818ec 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SlidingWindowSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/SlidingWindowSupplier.java
@@ -26,7 +26,6 @@
 import org.apache.kafka.streams.kstream.internals.FilteredIterator;
 import org.apache.kafka.streams.kstream.internals.WindowSupport;
 import org.apache.kafka.streams.processor.StateRestoreCallback;
-import org.apache.kafka.streams.processor.internals.ProcessorContextImpl;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.internals.Stamped;
@@ -186,7 +185,7 @@ public void flush() {
             IntegerSerializer intSerializer = new IntegerSerializer();
             ByteArraySerializer byteArraySerializer = new ByteArraySerializer();
 
-            RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
+            RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector();
 
             for (Map.Entry<K, ValueList<V>> entry : map.entrySet()) {
                 ValueList<V> values = entry.getValue();
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
index 60ac1df98b241..5cb53a41831fc 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
@@ -37,7 +37,7 @@
 import java.util.Map;
 import java.util.Set;
 
-public class ProcessorContextImpl implements ProcessorContext {
+public class ProcessorContextImpl implements ProcessorContext, RecordCollector.Supplier {
 
     private static final Logger log = LoggerFactory.getLogger(ProcessorContextImpl.class);
 
@@ -75,6 +75,7 @@ public ProcessorContextImpl(int id,
         this.initialized = false;
     }
 
+    @Override
     public RecordCollector recordCollector() {
         return this.collector;
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
index ad2f647a2ac3d..f0dbf35d9545f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
@@ -31,6 +31,17 @@
 
 public class RecordCollector {
 
+    /**
+     * A supplier of a {@link RecordCollector} instance.
+     */
+    public static interface Supplier {
+        /**
+         * Get the record collector.
+         * @return the record collector
+         */
+        public RecordCollector recordCollector();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(RecordCollector.class);
 
     private final Producer<byte[], byte[]> producer;
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
index e2d881c9cdea8..9f017272050e6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -53,7 +53,7 @@ public void init(ProcessorContext context) {
     @Override
     public void process(K key, V value) {
         // send to all the registered topics
-        RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
+        RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector();
         collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer);
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
index f45594e3d3fbd..90eee054371cd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
@@ -25,7 +25,6 @@
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.streams.processor.internals.ProcessorContextImpl;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
 
 import java.util.HashSet;
@@ -191,7 +190,7 @@ public void flush() {
     }
 
     private void logChange() {
-        RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
+        RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector();
         if (collector != null) {
             Serializer<K> keySerializer = serialization.keySerializer();
             Serializer<V> valueSerializer = serialization.valueSerializer();
diff --git a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
index c0b09f622986b..761f5ce902ef1 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java
@@ -23,31 +23,65 @@
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
 
 import java.io.File;
 import java.util.HashMap;
 import java.util.Map;
 
-public class MockProcessorContext implements ProcessorContext {
+public class MockProcessorContext implements ProcessorContext, RecordCollector.Supplier {
 
     private final KStreamTestDriver driver;
-    private final Serializer serializer;
-    private final Deserializer deserializer;
+    private final Serializer keySerializer;
+    private final Serializer valueSerializer;
+    private final Deserializer keyDeserializer;
+    private final Deserializer valueDeserializer;
+    private final RecordCollector.Supplier recordCollectorSupplier;
 
     private Map<String, StateStore> storeMap = new HashMap<>();
 
     long timestamp = -1L;
 
     public MockProcessorContext(KStreamTestDriver driver, Serializer<?> serializer, Deserializer<?> deserializer) {
+        this(driver, serializer, deserializer, serializer, deserializer, (RecordCollector.Supplier) null);
+    }
+
+    public MockProcessorContext(KStreamTestDriver driver, Serializer<?> keySerializer, Deserializer<?> keyDeserializer,
+            Serializer<?> valueSerializer, Deserializer<?> valueDeserializer,
+            final RecordCollector collector) {
+        this(driver, keySerializer, keyDeserializer, valueSerializer, valueDeserializer,
+                collector == null ? null : new RecordCollector.Supplier() {
+                    @Override
+                    public RecordCollector recordCollector() {
+                        return collector;
+                    }
+                });
+    }
+
+    public MockProcessorContext(KStreamTestDriver driver, Serializer<?> keySerializer, Deserializer<?> keyDeserializer,
+            Serializer<?> valueSerializer, Deserializer<?> valueDeserializer,
+            RecordCollector.Supplier collectorSupplier) {
         this.driver = driver;
-        this.serializer = serializer;
-        this.deserializer = deserializer;
+        this.keySerializer = keySerializer;
+        this.valueSerializer = valueSerializer;
+        this.keyDeserializer = keyDeserializer;
+        this.valueDeserializer = valueDeserializer;
+        this.recordCollectorSupplier = collectorSupplier;
+    }
+
+    @Override
+    public RecordCollector recordCollector() {
+        if (recordCollectorSupplier == null) {
+            throw new UnsupportedOperationException("No RecordCollector specified");
+        }
+        return recordCollectorSupplier.recordCollector();
     }
 
     public void setTime(long timestamp) {
         this.timestamp = timestamp;
     }
 
+    @Override
     public int id() {
         return -1;
     }
@@ -59,22 +93,22 @@ public boolean joinable() {
 
     @Override
     public Serializer<?> keySerializer() {
-        return serializer;
+        return keySerializer;
     }
 
     @Override
     public Serializer<?> valueSerializer() {
-        return serializer;
+        return valueSerializer;
     }
 
     @Override
     public Deserializer<?> keyDeserializer() {
-        return deserializer;
+        return keyDeserializer;
     }
 
     @Override
     public Deserializer<?> valueDeserializer() {
-        return deserializer;
+        return valueDeserializer;
     }
 
     @Override
@@ -140,4 +174,4 @@ public long timestamp() {
         return this.timestamp;
     }
 
-}
+}
\ No newline at end of file

From 9d5e6e56512f88c09a2648f1fae342d0def22a3f Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Wed, 30 Sep 2015 16:59:36 -0500
Subject: [PATCH 3/6] KAFKA-2593: Corrected RocksDBKeyValueStore to properly
 create directory if missing

---
 .../org/apache/kafka/streams/state/RocksDBKeyValueStore.java     | 1 +
 1 file changed, 1 insertion(+)

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
index fbd7002c6db50..f2ea13f343ec1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
@@ -230,6 +230,7 @@ public RocksDBStore(String name, ProcessorContext context, Serdes<K, V> serdes)
         private RocksDB openDB(File dir, Options options, int ttl) {
             try {
                 if (ttl == TTL_NOT_USED) {
+                    dir.getParentFile().mkdirs();
                     return RocksDB.open(options, dir.toString());
                 } else {
                     throw new KafkaException("Change log is not supported for store " + this.topic + " since it is TTL based.");

From a327aac88b49b7926765df96d5c7de0e542c9307 Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Wed, 30 Sep 2015 17:05:54 -0500
Subject: [PATCH 4/6] KAFKA-2593 Added unit tests for the key value stores

Added unit tests for the existing InMemoryKeyValueStore and RocksDBKeyValueStore implementations. A new KeyValueStoreTestDriver class does most of the work and simplifies the tests.
---
 .../state/AbstractKeyValueStoreTest.java      | 230 ++++++++++
 .../state/InMemoryKeyValueStoreTest.java      |  31 ++
 .../state/KeyValueStoreTestDriver.java        | 407 ++++++++++++++++++
 .../state/RocksDBKeyValueStoreTest.java       |  31 ++
 4 files changed, 699 insertions(+)
 create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
 create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
 create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
 create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java

diff --git a/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
new file mode 100644
index 0000000000000..007b580aa41e0
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
@@ -0,0 +1,230 @@
+/**
+ * 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.
+ */
+
+package org.apache.kafka.streams.state;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.junit.After;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+
+public abstract class AbstractKeyValueStoreTest {
+
+    protected static final File STATE_DIR = new File("build/data").getAbsoluteFile();
+
+    protected abstract <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context,
+                                                                      Class<K> keyClass, Class<V> valueClass,
+                                                                      boolean useContextSerdes);
+
+    @After
+    public void cleanup() {
+        if (STATE_DIR.exists()) {
+            try {
+                Files.walkFileTree(STATE_DIR.toPath(), new SimpleFileVisitor<Path>() {
+                    @Override
+                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+                        Files.delete(file);
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                    @Override
+                    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
+                        Files.delete(dir);
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                });
+            } catch (IOException e) {
+                // do nothing
+            }
+        }
+    }
+
+    @Test
+    public void testIntegerKeysAndStringValues() {
+        // Create the test driver ...
+        KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create();
+        driver.useStateDir(STATE_DIR);
+        KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, false);
+        try {
+
+            // Verify that the store reads and writes correctly ...
+            store.put(0, "zero");
+            store.put(1, "one");
+            store.put(2, "two");
+            store.put(4, "four");
+            store.put(5, "five");
+            assertEquals(5, driver.sizeOf(store));
+            assertEquals("zero", store.get(0));
+            assertEquals("one", store.get(1));
+            assertEquals("two", store.get(2));
+            assertNull(store.get(3));
+            assertEquals("four", store.get(4));
+            assertEquals("five", store.get(5));
+            store.delete(5);
+
+            // Flush the store and verify all current entries were properly flushed ...
+            store.flush();
+            assertEquals("zero", driver.flushedEntryStored(0));
+            assertEquals("one", driver.flushedEntryStored(1));
+            assertEquals("two", driver.flushedEntryStored(2));
+            assertEquals("four", driver.flushedEntryStored(4));
+            assertEquals(null, driver.flushedEntryStored(5));
+
+            assertEquals(false, driver.flushedEntryRemoved(0));
+            assertEquals(false, driver.flushedEntryRemoved(1));
+            assertEquals(false, driver.flushedEntryRemoved(2));
+            assertEquals(false, driver.flushedEntryRemoved(4));
+            assertEquals(true, driver.flushedEntryRemoved(5));
+
+            // Check range iteration ...
+            try (KeyValueIterator<Integer, String> iter = store.range(2, 4)) {
+                while (iter.hasNext()) {
+                    Entry<Integer, String> entry = iter.next();
+                    if (entry.key().equals(2))
+                        assertEquals("two", entry.value());
+                    else if (entry.key().equals(4))
+                        assertEquals("four", entry.value());
+                    else
+                        fail("Unexpected entry: " + entry);
+                }
+            }
+
+            // Check range iteration ...
+            try (KeyValueIterator<Integer, String> iter = store.range(2, 6)) {
+                while (iter.hasNext()) {
+                    Entry<Integer, String> entry = iter.next();
+                    if (entry.key().equals(2))
+                        assertEquals("two", entry.value());
+                    else if (entry.key().equals(4))
+                        assertEquals("four", entry.value());
+                    else
+                        fail("Unexpected entry: " + entry);
+                }
+            }
+        } finally {
+            store.close();
+        }
+    }
+
+    @Test
+    public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializers() {
+        // Create the test driver ...
+        KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
+        driver.useStateDir(STATE_DIR);
+        KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, true);
+        try {
+
+            // Verify that the store reads and writes correctly ...
+            store.put(0, "zero");
+            store.put(1, "one");
+            store.put(2, "two");
+            store.put(4, "four");
+            store.put(5, "five");
+            assertEquals(5, driver.sizeOf(store));
+            assertEquals("zero", store.get(0));
+            assertEquals("one", store.get(1));
+            assertEquals("two", store.get(2));
+            assertNull(store.get(3));
+            assertEquals("four", store.get(4));
+            assertEquals("five", store.get(5));
+            store.delete(5);
+
+            // Flush the store and verify all current entries were properly flushed ...
+            store.flush();
+            assertEquals("zero", driver.flushedEntryStored(0));
+            assertEquals("one", driver.flushedEntryStored(1));
+            assertEquals("two", driver.flushedEntryStored(2));
+            assertEquals("four", driver.flushedEntryStored(4));
+            assertEquals(null, driver.flushedEntryStored(5));
+
+            assertEquals(false, driver.flushedEntryRemoved(0));
+            assertEquals(false, driver.flushedEntryRemoved(1));
+            assertEquals(false, driver.flushedEntryRemoved(2));
+            assertEquals(false, driver.flushedEntryRemoved(4));
+            assertEquals(true, driver.flushedEntryRemoved(5));
+        } finally {
+            store.close();
+        }
+    }
+
+    @Test
+    public void testRestoringInetgerKeysAndValues() {
+        // Create the test driver ...
+        KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
+        driver.useStateDir(STATE_DIR);
+
+        // Add any entries that will be restored to any store
+        // that uses the driver's context ...
+        driver.addRestoreEntry(0, "zero");
+        driver.addRestoreEntry(1, "one");
+        driver.addRestoreEntry(2, "two");
+        driver.addRestoreEntry(4, "four");
+
+        // Create the store, which should register with the context and automatically
+        // receive the restore entries ...
+        KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, false);
+        try {
+            // Verify that the store's contents were properly restored ...
+            assertEquals(0, driver.checkForRestoredEntries(store));
+
+            // and there are no other entries ...
+            assertEquals(4, driver.sizeOf(store));
+        } finally {
+            store.close();
+        }
+    }
+
+    @Test
+    public void testRestoringInetgerKeysAndValuesUsingDefaultSerializersAndDeserializers() {
+        // Create the test driver ...
+        KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
+        driver.useStateDir(STATE_DIR);
+
+        // Add any entries that will be restored to any store
+        // that uses the driver's context ...
+        driver.addRestoreEntry(0, "zero");
+        driver.addRestoreEntry(1, "one");
+        driver.addRestoreEntry(2, "two");
+        driver.addRestoreEntry(4, "four");
+
+        // Create the store, which should register with the context and automatically
+        // receive the restore entries ...
+        KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, true);
+        try {
+            // Verify that the store's contents were properly restored ...
+            assertEquals(0, driver.checkForRestoredEntries(store));
+
+            // and there are no other entries ...
+            assertEquals(4, driver.sizeOf(store));
+        } finally {
+            store.close();
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
new file mode 100644
index 0000000000000..9ca42b9f1d4a6
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
@@ -0,0 +1,31 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+public class InMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest {
+
+    @Override
+    protected <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
+                                                             boolean useContextSerdes) {
+        if (useContextSerdes) {
+            return InMemoryKeyValueStore.create("my-store", context);
+        }
+        return InMemoryKeyValueStore.create("my-store", context, keyClass, valueClass);
+    }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
new file mode 100644
index 0000000000000..5a47a29d66ca7
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -0,0 +1,407 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.RestoreFunc;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.RecordCollector;
+import org.apache.kafka.test.MockProcessorContext;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A component that provides a {@link #context() ProcessingContext} that can be supplied to a {@link KeyValueStore} so that
+ * all entries written to the Kafka topic by the store during {@link KeyValueStore#flush()} are captured for testing purposes.
+ * This class simplifies testing of various {@link KeyValueStore} instances, especially those that use
+ * {@link MeteredKeyValueStore} to monitor and write its entries to the Kafka topic.
+ * <p>
+ * <h2>Basic usage</h2>
+ * This component can be used to help test a {@link KeyValueStore}'s ability to read and write entries.
+ * 
+ * <pre>
+ * &#47;&#47; Create the test driver ...
+ * KeyValueStoreTestDriver&lt;Integer, String> driver = KeyValueStoreTestDriver.create();
+ * InMemoryKeyValueStore&lt;Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
+ *                                                                             Integer.class, String.class);
+ * 
+ * &#47;&#47; Verify that the store reads and writes correctly ...
+ * store.put(0, "zero");
+ * store.put(1, "one");
+ * store.put(2, "two");
+ * store.put(4, "four");
+ * store.put(5, "five");
+ * assertEquals(5, driver.sizeOf(store));
+ * assertEquals("zero", store.get(0));
+ * assertEquals("one", store.get(1));
+ * assertEquals("two", store.get(2));
+ * assertEquals("four", store.get(4));
+ * assertEquals("five", store.get(5));
+ * assertNull(store.get(3));
+ * store.delete(5);
+ * 
+ * &#47;&#47; Flush the store and verify all current entries were properly flushed ...
+ * store.flush();
+ * assertEquals("zero", driver.flushedEntryStored(0));
+ * assertEquals("one", driver.flushedEntryStored(1));
+ * assertEquals("two", driver.flushedEntryStored(2));
+ * assertEquals("four", driver.flushedEntryStored(4));
+ * assertEquals(null, driver.flushedEntryStored(5));
+ * 
+ * assertEquals(false, driver.flushedEntryRemoved(0));
+ * assertEquals(false, driver.flushedEntryRemoved(1));
+ * assertEquals(false, driver.flushedEntryRemoved(2));
+ * assertEquals(false, driver.flushedEntryRemoved(4));
+ * assertEquals(true, driver.flushedEntryRemoved(5));
+ * </pre>
+ * 
+ * <p>
+ * <h2>Restoring a store</h2>
+ * This component can be used to test whether a {@link KeyValueStore} implementation properly
+ * {@link ProcessorContext#register(StateStore, RestoreFunc) registers itself} with the {@link ProcessorContext}. To do this,
+ * create an instance of this driver component, {@link #addRestoreEntry(Object, Object) add entries} that will be passed to the
+ * store upon creation, and then create the store using this driver's {@link #context() ProcessorContext}:
+ * 
+ * <pre>
+ * &#47;&#47; Create the test driver ...
+ * KeyValueStoreTestDriver&lt;Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
+ * 
+ * &#47;&#47; Add any entries that will be restored to any store
+ * &#47;&#47; that uses the driver's context ...
+ * driver.addRestoreEntry(0, "zero");
+ * driver.addRestoreEntry(1, "one");
+ * driver.addRestoreEntry(2, "two");
+ * driver.addRestoreEntry(4, "four");
+ * 
+ * &#47;&#47; Create the store, which should register with the context and automatically
+ * &#47;&#47; receive the restore entries ...
+ * InMemoryKeyValueStore&lt;Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
+ *                                                                             Integer.class, String.class);
+ * 
+ * &#47;&#47; Verify that the store's contents were properly restored ...
+ * assertEquals(0, driver.checkForRestoredEntries(store));
+ * 
+ * &#47;&#47; and there are no other entries ...
+ * assertEquals(4, driver.sizeOf(store));
+ * </pre>
+ * 
+ * @param <K> the type of keys placed in the store
+ * @param <V> the type of values placed in the store
+ */
+public class KeyValueStoreTestDriver<K, V> {
+
+    private static <T> Serializer<T> unusableSerializer() {
+        return new Serializer<T>() {
+            @Override
+            public void configure(Map<String, ?> configs, boolean isKey) {
+            }
+
+            @Override
+            public byte[] serialize(String topic, T data) {
+                throw new UnsupportedOperationException("This serializer should not be used");
+            }
+
+            @Override
+            public void close() {
+            }
+        };
+    };
+
+    private static <T> Deserializer<T> unusableDeserializer() {
+        return new Deserializer<T>() {
+            @Override
+            public void configure(Map<String, ?> configs, boolean isKey) {
+            }
+
+            @Override
+            public T deserialize(String topic, byte[] data) {
+                throw new UnsupportedOperationException("This serializer should not be used");
+            }
+
+            @Override
+            public void close() {
+            }
+        };
+    };
+
+    /**
+     * Create a driver object that will have a {@link #context()} that records messages
+     * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides <em>unusable</em> default key and
+     * value serializers and deserializers. This can be used when the actual serializers and deserializers are supplied to the
+     * store during creation, which should eliminate the need for a store to depend on the ProcessorContext's default key and
+     * value serializers and deserializers.
+     * 
+     * @return the test driver; never null
+     */
+    public static <K, V> KeyValueStoreTestDriver<K, V> create() {
+        Serializer<K> keySerializer = unusableSerializer();
+        Deserializer<K> keyDeserializer = unusableDeserializer();
+        Serializer<V> valueSerializer = unusableSerializer();
+        Deserializer<V> valueDeserializer = unusableDeserializer();
+        Serdes<K, V> serdes = new Serdes<K, V>("unexpected", keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
+        return new KeyValueStoreTestDriver<K, V>(serdes);
+    }
+
+    /**
+     * Create a driver object that will have a {@link #context()} that records messages
+     * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides default serializers and
+     * deserializers for the given built-in key and value types (e.g., {@code String.class}, {@code Integer.class},
+     * {@code Long.class}, and {@code byte[].class}). This can be used when store is created to rely upon the
+     * ProcessorContext's default key and value serializers and deserializers.
+     * 
+     * @param keyClass the class for the keys; must be one of {@code String.class}, {@code Integer.class},
+     *            {@code Long.class}, or {@code byte[].class}
+     * @param valueClass the class for the values; must be one of {@code String.class}, {@code Integer.class},
+     *            {@code Long.class}, or {@code byte[].class}
+     * @return the test driver; never null
+     */
+    public static <K, V> KeyValueStoreTestDriver<K, V> create(Class<K> keyClass, Class<V> valueClass) {
+        Serdes<K, V> serdes = Serdes.withBuiltinTypes("unexpected", keyClass, valueClass);
+        return new KeyValueStoreTestDriver<K, V>(serdes);
+    }
+
+    /**
+     * Create a driver object that will have a {@link #context()} that records messages
+     * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides the specified serializers and
+     * deserializers. This can be used when store is created to rely upon the ProcessorContext's default key and value serializers
+     * and deserializers.
+     * 
+     * @param keySerializer the key serializer for the {@link ProcessorContext}; may not be null
+     * @param keyDeserializer the key deserializer for the {@link ProcessorContext}; may not be null
+     * @param valueSerializer the value serializer for the {@link ProcessorContext}; may not be null
+     * @param valueDeserializer the value deserializer for the {@link ProcessorContext}; may not be null
+     * @return the test driver; never null
+     */
+    public static <K, V> KeyValueStoreTestDriver<K, V> create(Serializer<K> keySerializer,
+                                                              Deserializer<K> keyDeserializer,
+                                                              Serializer<V> valueSerializer,
+                                                              Deserializer<V> valueDeserializer) {
+        Serdes<K, V> serdes = new Serdes<K, V>("unexpected", keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
+        return new KeyValueStoreTestDriver<K, V>(serdes);
+    }
+
+    private final Serdes<K, V> serdes;
+    private final Map<K, V> flushedEntries = new HashMap<>();
+    private final Set<K> flushedRemovals = new HashSet<>();
+    private final List<Entry<K, V>> restorableEntries = new LinkedList<>();
+    private final MockProcessorContext context;
+    private final Map<String, StateStore> storeMap = new HashMap<>();
+    private final Metrics metrics = new Metrics();
+    private final RecordCollector recordCollector;
+    private File stateDir = new File("build/data").getAbsoluteFile();
+
+    protected KeyValueStoreTestDriver(Serdes<K, V> serdes) {
+        this.serdes = serdes;
+        ByteArraySerializer rawSerializer = new ByteArraySerializer();
+        Producer<byte[], byte[]> producer = new MockProducer<byte[], byte[]>(true, rawSerializer, rawSerializer);
+        this.recordCollector = new RecordCollector(producer) {
+            @Override
+            public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) {
+                recordFlushed(record.key(), record.value());
+            }
+        };
+        this.context = new MockProcessorContext(null, serdes.keySerializer(), serdes.keyDeserializer(), serdes.valueSerializer(),
+                serdes.valueDeserializer(), recordCollector) {
+            @Override
+            public int id() {
+                return 1;
+            }
+
+            @Override
+            public <K1, V1> void forward(K1 key, V1 value, int childIndex) {
+                forward(key, value);
+            }
+
+            @Override
+            public void register(StateStore store, RestoreFunc func) {
+                storeMap.put(store.name(), store);
+                restoreEntries(func);
+            }
+
+            @Override
+            public StateStore getStateStore(String name) {
+                return storeMap.get(name);
+            }
+
+            @Override
+            public Metrics metrics() {
+                return metrics;
+            }
+
+            @Override
+            public File stateDir() {
+                if (stateDir == null) {
+                    throw new UnsupportedOperationException("No state directory set");
+                }
+                stateDir.mkdirs();
+                return stateDir;
+            }
+        };
+    }
+
+    /**
+     * Set the directory that should be used by the store for local disk storage.
+     * 
+     * @param dir the directory; may be null if no local storage is allowed
+     */
+    public void useStateDir(File dir) {
+        this.stateDir = dir;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected <K1, V1> void recordFlushed(K1 key, V1 value) {
+        K k = (K) key;
+        if (value == null) {
+            // This is a removal ...
+            flushedRemovals.add(k);
+            flushedEntries.remove(k);
+        } else {
+            // This is a normal add
+            flushedEntries.put(k, (V) value);
+            flushedRemovals.remove(k);
+        }
+    }
+
+    private void restoreEntries(RestoreFunc func) {
+        for (Entry<K, V> entry : restorableEntries) {
+            if (entry != null) {
+                byte[] rawKey = serdes.rawKey(entry.key());
+                byte[] rawValue = serdes.rawValue(entry.value());
+                func.apply(rawKey, rawValue);
+            }
+        }
+    }
+
+    /**
+     * This method adds an entry to the "restore log" for the {@link KeyValueStore}. This should be called <em>before</em>
+     * creating
+     * the {@link KeyValueStore} with the {@link #context() ProcessorContext}; when the {@link KeyValueStore} is created, it will
+     * {@link ProcessorContext#register(StateStore, RestoreFunc) register} itself with the {@link #context() ProcessorContext},
+     * and this object will then pre-populate the store with all restore entries added via this method.
+     * 
+     * @param key the key for the entry
+     * @param value the value for the entry
+     */
+    public void addRestoreEntry(K key, V value) {
+        restorableEntries.add(new Entry<K, V>(key, value));
+    }
+
+    /**
+     * Get the context that should be supplied to a {@link KeyValueStore}'s constructor. This context records any messages
+     * written by the store to the Kafka topic, making them available via the {@link #flushedEntryStored(Object)} and
+     * {@link #flushedEntryRemoved(Object)} methods.
+     * <p>
+     * If the {@link KeyValueStore}'s are to be restored upon its startup, be sure to {@link #addRestoreEntry(Object, Object)
+     * add the restore entries} before creating the store with the {@link ProcessorContext} returned by this method.
+     * 
+     * @return the processing context; never null
+     * @see #addRestoreEntry(Object, Object)
+     */
+    public ProcessorContext context() {
+        return context;
+    }
+
+    /**
+     * Get the entries that are restored to a KeyValueStore when it is constructed with this driver's {@link #context()
+     * ProcessorContext}.
+     * 
+     * @return the restore entries; never null but possibly a null iterator
+     */
+    public Iterable<Entry<K, V>> restoredEntries() {
+        return restorableEntries;
+    }
+
+    /**
+     * Utility method that will count the number of {@link #addRestoreEntry(Object, Object) restore entries} missing from the
+     * supplied store.
+     * 
+     * @param store the store that is to have all of the {@link #restoredEntries() restore entries}
+     * @return the number of restore entries missing from the store, or 0 if all restore entries were found
+     */
+    public int checkForRestoredEntries(KeyValueStore<K, V> store) {
+        int missing = 0;
+        for (Entry<K, V> entry : restorableEntries) {
+            if (entry != null) {
+                V value = store.get(entry.key());
+                if (!Objects.equals(value, entry.value())) {
+                    ++missing;
+                }
+            }
+        }
+        return missing;
+    }
+
+    /**
+     * Utility method to compute the number of entries within the store.
+     * 
+     * @param store the key value store using this {@link #context()}.
+     * @return the number of entries
+     */
+    public int sizeOf(KeyValueStore<K, V> store) {
+        int size = 0;
+        for (KeyValueIterator<K, V> iterator = store.all(); iterator.hasNext();) {
+            iterator.next();
+            ++size;
+        }
+        return size;
+    }
+
+    /**
+     * Retrieve the value that the store {@link KeyValueStore#flush() flushed} with the given key.
+     * 
+     * @param key the key
+     * @return the value that was flushed with the key, or {@code null} if no such key was flushed or if the entry with this
+     *         key was {@link #flushedEntryStored(Object) removed} upon flush
+     */
+    public V flushedEntryStored(K key) {
+        return flushedEntries.get(key);
+    }
+
+    /**
+     * Determine whether the store {@link KeyValueStore#flush() flushed} the removal of the given key.
+     * 
+     * @param key the key
+     * @return {@code true} if the entry with the given key was removed when flushed, or {@code false} if the entry was not
+     *         removed when last flushed
+     */
+    public boolean flushedEntryRemoved(K key) {
+        return flushedRemovals.contains(key);
+    }
+
+    /**
+     * Remove all {@link #flushedEntryStored(Object) flushed entries}, {@link #flushedEntryRemoved(Object) flushed removals},
+     */
+    public void clear() {
+        restorableEntries.clear();
+        flushedEntries.clear();
+        flushedRemovals.clear();
+    }
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java
new file mode 100644
index 0000000000000..d0c00fdb6477d
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java
@@ -0,0 +1,31 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+public class RocksDBKeyValueStoreTest extends AbstractKeyValueStoreTest {
+
+    @Override
+    protected <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
+                                                             boolean useContextSerdes) {
+        if (useContextSerdes) {
+            return RocksDBKeyValueStore.create("my-store", context);
+        }
+        return RocksDBKeyValueStore.create("my-store", context, keyClass, valueClass);
+    }
+}
\ No newline at end of file

From 5b107f236b10efea1b136c657c90677bd2bdd36a Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Wed, 30 Sep 2015 17:36:46 -0500
Subject: [PATCH 5/6] KAFKA-2593 Correct how Kafka Streams tests create and use
 local state directory

The ProcessorTopologyTest and AbstractKeyValueStoreTest test classes have methods created a shared directory used by all their tests to store local state. When the computer has multiple processors, the build system executes test in parallel, and they interfere with other tests using the same directory. Solve the problem by adding a utility class to create temporary directories and automatically clean them up.
---
 .../internals/ProcessorTopologyTest.java      | 52 ++++---------
 .../state/AbstractKeyValueStoreTest.java      | 41 +---------
 .../state/KeyValueStoreTestDriver.java        | 25 ++++--
 .../kafka/streams/state/StateUtils.java       | 77 +++++++++++++++++++
 4 files changed, 111 insertions(+), 84 deletions(-)
 create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/StateUtils.java

diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index c44fe13e902b9..dd60ad4878cb0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -37,6 +37,7 @@
 import org.apache.kafka.streams.state.InMemoryKeyValueStore;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StateUtils;
 import org.apache.kafka.test.MockProcessorSupplier;
 import org.apache.kafka.test.ProcessorTopologyTestDriver;
 import org.junit.After;
@@ -44,19 +45,12 @@
 import org.junit.Test;
 
 import java.io.File;
-import java.io.IOException;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Properties;
 
 public class ProcessorTopologyTest {
 
     private static final Serializer<String> STRING_SERIALIZER = new StringSerializer();
     private static final Deserializer<String> STRING_DESERIALIZER = new StringDeserializer();
-    private static final File STATE_DIR = new File("build/data").getAbsoluteFile();
 
     protected static final String INPUT_TOPIC = "input-topic";
     protected static final String OUTPUT_TOPIC_1 = "output-topic-1";
@@ -69,10 +63,11 @@ public class ProcessorTopologyTest {
 
     @Before
     public void setup() {
-        STATE_DIR.mkdirs();
+        // Create a new directory in which we'll put all of the state for this test, enabling running tests in parallel ...
+        File localState = StateUtils.tempDir();
         Properties props = new Properties();
         props.setProperty(StreamingConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
-        props.setProperty(StreamingConfig.STATE_DIR_CONFIG, STATE_DIR.getAbsolutePath());
+        props.setProperty(StreamingConfig.STATE_DIR_CONFIG, localState.getAbsolutePath());
         props.setProperty(StreamingConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, CustomTimestampExtractor.class.getName());
         props.setProperty(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.setProperty(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
@@ -87,36 +82,16 @@ public void cleanup() {
             driver.close();
         }
         driver = null;
-        if (STATE_DIR.exists()) {
-            try {
-                Files.walkFileTree(STATE_DIR.toPath(), new SimpleFileVisitor<Path>() {
-                    @Override
-                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
-                        Files.delete(file);
-                        return FileVisitResult.CONTINUE;
-                    }
-    
-                    @Override
-                    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
-                        Files.delete(dir);
-                        return FileVisitResult.CONTINUE;
-                    }
-    
-                });
-            } catch (IOException e) {
-                // do nothing
-            }
-        }
     }
-
+    
     @Test
     public void testTopologyMetadata() {
         final TopologyBuilder builder = new TopologyBuilder();
 
         builder.addSource("source-1", "topic-1");
         builder.addSource("source-2", "topic-2", "topic-3");
-        builder.addProcessor("processor-1", new MockProcessorSupplier(), "source-1");
-        builder.addProcessor("processor-2", new MockProcessorSupplier(), "source-1", "source-2");
+        builder.addProcessor("processor-1", new MockProcessorSupplier<>(), "source-1");
+        builder.addProcessor("processor-2", new MockProcessorSupplier<>(), "source-1", "source-2");
         builder.addSink("sink-1", "topic-3", "processor-1");
         builder.addSink("sink-2", "topic-4", "processor-1", "processor-2");
 
@@ -306,12 +281,17 @@ public void punctuate(long streamTime) {
             }
             context().forward(Long.toString(streamTime), count);
         }
+        
+        @Override
+        public void close() {
+            store.close();
+        }
     }
 
-    protected ProcessorSupplier define(final Processor processor) {
-        return new ProcessorSupplier() {
+    protected <K, V> ProcessorSupplier<K, V> define(final Processor<K, V> processor) {
+        return new ProcessorSupplier<K, V>() {
             @Override
-            public Processor get() {
+            public Processor<K, V> get() {
                 return processor;
             }
         };
@@ -323,4 +303,4 @@ public long extract(ConsumerRecord<Object, Object> record) {
             return timestamp;
         }
     }
-}
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
index 007b580aa41e0..9a7b029412252 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
@@ -22,54 +22,18 @@
 import static org.junit.Assert.fail;
 
 import org.apache.kafka.streams.processor.ProcessorContext;
-import org.junit.After;
 import org.junit.Test;
 
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
-
 public abstract class AbstractKeyValueStoreTest {
 
-    protected static final File STATE_DIR = new File("build/data").getAbsoluteFile();
-
     protected abstract <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context,
                                                                       Class<K> keyClass, Class<V> valueClass,
                                                                       boolean useContextSerdes);
-
-    @After
-    public void cleanup() {
-        if (STATE_DIR.exists()) {
-            try {
-                Files.walkFileTree(STATE_DIR.toPath(), new SimpleFileVisitor<Path>() {
-                    @Override
-                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
-                        Files.delete(file);
-                        return FileVisitResult.CONTINUE;
-                    }
-
-                    @Override
-                    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
-                        Files.delete(dir);
-                        return FileVisitResult.CONTINUE;
-                    }
-
-                });
-            } catch (IOException e) {
-                // do nothing
-            }
-        }
-    }
-
+    
     @Test
     public void testIntegerKeysAndStringValues() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create();
-        driver.useStateDir(STATE_DIR);
         KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, false);
         try {
 
@@ -136,7 +100,6 @@ else if (entry.key().equals(4))
     public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializers() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
-        driver.useStateDir(STATE_DIR);
         KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, true);
         try {
 
@@ -177,7 +140,6 @@ public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializer
     public void testRestoringInetgerKeysAndValues() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
-        driver.useStateDir(STATE_DIR);
 
         // Add any entries that will be restored to any store
         // that uses the driver's context ...
@@ -204,7 +166,6 @@ public void testRestoringInetgerKeysAndValues() {
     public void testRestoringInetgerKeysAndValuesUsingDefaultSerializersAndDeserializers() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
-        driver.useStateDir(STATE_DIR);
 
         // Add any entries that will be restored to any store
         // that uses the driver's context ...
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 5a47a29d66ca7..25652eac20e43 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -19,12 +19,13 @@
 import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.StreamingMetrics;
 import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.RestoreFunc;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
 import org.apache.kafka.test.MockProcessorContext;
@@ -214,7 +215,15 @@ public static <K, V> KeyValueStoreTestDriver<K, V> create(Serializer<K> keySeria
     private final List<Entry<K, V>> restorableEntries = new LinkedList<>();
     private final MockProcessorContext context;
     private final Map<String, StateStore> storeMap = new HashMap<>();
-    private final Metrics metrics = new Metrics();
+    private final StreamingMetrics metrics = new StreamingMetrics() {
+        @Override
+        public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
+            return null;
+        }
+        @Override
+        public void recordLatency(Sensor sensor, long startNs, long endNs) {
+        }
+    };
     private final RecordCollector recordCollector;
     private File stateDir = new File("build/data").getAbsoluteFile();
 
@@ -241,7 +250,7 @@ public <K1, V1> void forward(K1 key, V1 value, int childIndex) {
             }
 
             @Override
-            public void register(StateStore store, RestoreFunc func) {
+            public void register(StateStore store, StateRestoreCallback func) {
                 storeMap.put(store.name(), store);
                 restoreEntries(func);
             }
@@ -252,14 +261,14 @@ public StateStore getStateStore(String name) {
             }
 
             @Override
-            public Metrics metrics() {
+            public StreamingMetrics metrics() {
                 return metrics;
             }
 
             @Override
             public File stateDir() {
                 if (stateDir == null) {
-                    throw new UnsupportedOperationException("No state directory set");
+                    stateDir = StateUtils.tempDir();
                 }
                 stateDir.mkdirs();
                 return stateDir;
@@ -290,12 +299,12 @@ protected <K1, V1> void recordFlushed(K1 key, V1 value) {
         }
     }
 
-    private void restoreEntries(RestoreFunc func) {
+    private void restoreEntries(StateRestoreCallback func) {
         for (Entry<K, V> entry : restorableEntries) {
             if (entry != null) {
                 byte[] rawKey = serdes.rawKey(entry.key());
                 byte[] rawValue = serdes.rawValue(entry.value());
-                func.apply(rawKey, rawValue);
+                func.restore(rawKey, rawValue);
             }
         }
     }
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/StateUtils.java b/streams/src/test/java/org/apache/kafka/streams/state/StateUtils.java
new file mode 100644
index 0000000000000..c7ea748ca7c47
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/StateUtils.java
@@ -0,0 +1,77 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.test.TestUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A utility for tests to create and manage unique and isolated directories on the file system for local state.
+ */
+public class StateUtils {
+
+    private static final AtomicLong INSTANCE_COUNTER = new AtomicLong();
+
+    /**
+     * Create a new temporary directory that will be cleaned up automatically upon shutdown.
+     * @return the new directory that will exist; never null
+     */
+    public static File tempDir() {
+        final File dir = new File(TestUtils.IO_TMP_DIR, "kafka-" + INSTANCE_COUNTER.incrementAndGet());
+        dir.mkdirs();
+        dir.deleteOnExit();
+
+        Runtime.getRuntime().addShutdownHook(new Thread() {
+            @Override
+            public void run() {
+                deleteDirectory(dir);
+            }
+        });
+        return dir;
+    }
+
+    private static void deleteDirectory(File dir) {
+        if (dir != null && dir.exists()) {
+            try {
+                Files.walkFileTree(dir.toPath(), new SimpleFileVisitor<Path>() {
+                    @Override
+                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+                        Files.delete(file);
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                    @Override
+                    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
+                        Files.delete(dir);
+                        return FileVisitResult.CONTINUE;
+                    }
+
+                });
+            } catch (IOException e) {
+                // do nothing
+            }
+        }
+    }
+}
\ No newline at end of file

From 05e9cd1aeb10c36cfb2dc4787ba9d3fa120e108a Mon Sep 17 00:00:00 2001
From: Randall Hauch <rhauch@gmail.com>
Date: Mon, 12 Oct 2015 09:48:55 -0500
Subject: [PATCH 6/6] KAFKA-2593 Added a fluent builder API for creating
 key-value stores

Added a simple fluent builder API to create key-value stores that use in-memory or a local database. Other kinds of state stores can be added in the future.
---
 .../kafka/streams/examples/ProcessorJob.java  |  12 +-
 .../streams/state/InMemoryKeyValueStore.java  | 120 +-------
 .../streams/state/RocksDBKeyValueStore.java   | 126 +--------
 .../apache/kafka/streams/state/Serdes.java    |   4 +-
 .../apache/kafka/streams/state/Stores.java    | 257 ++++++++++++++++++
 .../internals/ProcessorTopologyTest.java      |   4 +-
 .../state/AbstractKeyValueStoreTest.java      |  24 +-
 .../state/InMemoryKeyValueStoreTest.java      |  11 +-
 .../state/KeyValueStoreTestDriver.java        |  79 ++++--
 .../state/RocksDBKeyValueStoreTest.java       |  11 +-
 10 files changed, 357 insertions(+), 291 deletions(-)
 create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/Stores.java

diff --git a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
index cdfc29b0e93be..0317b9d2d0d76 100644
--- a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
+++ b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java
@@ -17,20 +17,20 @@
 
 package org.apache.kafka.streams.examples;
 
-import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.streams.KafkaStreaming;
+import org.apache.kafka.streams.StreamingConfig;
 import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
 import org.apache.kafka.streams.processor.TopologyBuilder;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.StreamingConfig;
 import org.apache.kafka.streams.state.Entry;
-import org.apache.kafka.streams.state.InMemoryKeyValueStore;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.Stores;
 
 import java.util.Properties;
 
@@ -48,7 +48,7 @@ public Processor<String, String> get() {
                 public void init(ProcessorContext context) {
                     this.context = context;
                     this.context.schedule(1000);
-                    this.kvStore = InMemoryKeyValueStore.create("local-state", context, String.class, Integer.class);
+                    this.kvStore = Stores.create("local-state", context).withStringKeys().withIntegerValues().inMemory().build();
                 }
 
                 @Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
index 01cc53f68000d..1eb526f066794 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java
@@ -17,8 +17,6 @@
 
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.streams.processor.ProcessorContext;
@@ -30,127 +28,15 @@
 import java.util.TreeMap;
 
 /**
- * An in-memory key-value store based on a TreeMap
+ * An in-memory key-value store based on a TreeMap.
  *
  * @param <K> The key type
  * @param <V> The value type
+ * 
+ * @see Stores#create(String, ProcessorContext)
  */
 public class InMemoryKeyValueStore<K, V> extends MeteredKeyValueStore<K, V> {
 
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic and the system time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @return the key-value store
-     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
-     *             {@code Integer.class}, {@code Long.class}, or
-     *             {@code byte[].class}
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass) {
-        return create(name, context, keyClass, valueClass, new SystemTime());
-    }
-
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic and the given time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
-     *             {@code Integer.class}, {@code Long.class}, or
-     *             {@code byte[].class}
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
-                                                            Time time) {
-        return new InMemoryKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), time);
-    }
-
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default
-     * serializers and deserializers, and the system time provider.
-     * <p>
-     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
-     * used as parameters for this key value store. This is not checked in this method, and any mismatch will result in
-     * class cast exceptions during usage.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @return the key-value store
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context) {
-        return create(name, context, new SystemTime());
-    }
-
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default
-     * serializers and deserializers, and the given time provider.
-     * <p>
-     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
-     * used as parameters for this key value store.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Time time) {
-        Serdes<K, V> serdes = new Serdes<>(name, context);
-        return new InMemoryKeyValueStore<>(name, context, serdes, time);
-    }
-
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and
-     * the system time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keySerializer the serializer for keys; may not be null
-     * @param keyDeserializer the deserializer for keys; may not be null
-     * @param valueSerializer the serializer for values; may not be null
-     * @param valueDeserializer the deserializer for values; may not be null
-     * @return the key-value store
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context,
-                                                            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
-                                                            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) {
-        return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime());
-    }
-
-    /**
-     * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and
-     * the given time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keySerializer the serializer for keys; may not be null
-     * @param keyDeserializer the deserializer for keys; may not be null
-     * @param valueSerializer the serializer for values; may not be null
-     * @param valueDeserializer the deserializer for values; may not be null
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     */
-    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context,
-                                                            Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
-                                                            Serializer<V> valueSerializer, Deserializer<V> valueDeserializer,
-                                                            Time time) {
-        Serdes<K, V> serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
-        return new InMemoryKeyValueStore<>(name, context, serdes, time);
-    }
-
     protected InMemoryKeyValueStore(String name, ProcessorContext context, Serdes<K, V> serdes, Time time) {
         super(name, new MemoryStore<K, V>(name), context, serdes, "in-memory-state", time != null ? time : new SystemTime());
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
index f2ea13f343ec1..32897ea5e79df 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java
@@ -18,8 +18,6 @@
 package org.apache.kafka.streams.state;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.streams.processor.ProcessorContext;
@@ -43,125 +41,11 @@
  *
  * @param <K> the type of keys
  * @param <V> the type of values
+ * 
+ * @see Stores#create(String, ProcessorContext)
  */
 public class RocksDBKeyValueStore<K, V> extends MeteredKeyValueStore<K, V> {
 
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the system time
-     * provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @return the key-value store
-     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
-     *             {@code Integer.class}, {@code Long.class}, or
-     *             {@code byte[].class}
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass) {
-        return create(name, context, keyClass, valueClass, new SystemTime());
-    }
-
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the given time
-     * provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
-     *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
-     *            {@code byte[].class})
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     * @throws IllegalArgumentException if the {@code keyClass} or {@code valueClass} are not one of {@code String.class},
-     *             {@code Integer.class}, {@code Long.class}, or
-     *             {@code byte[].class}
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
-                                                           Time time) {
-        return new RocksDBKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), time);
-    }
-
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
-     * {@link ProcessorContext}'s default serializers and deserializers, and the system time provider.
-     * <p>
-     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
-     * used as parameters for this key value store. This is not checked in this method, and any mismatch will result in
-     * class cast exceptions during usage.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @return the key-value store
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context) {
-        return create(name, context, new SystemTime());
-    }
-
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
-     * {@link ProcessorContext}'s default serializers and deserializers, and the given time provider.
-     * <p>
-     * <strong>NOTE:</strong> the default serializers and deserializers in the context <em>must</em> match the key and value types
-     * used as parameters for this key value store.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context, Time time) {
-        Serdes<K, V> serdes = new Serdes<>(name, context);
-        return new RocksDBKeyValueStore<>(name, context, serdes, time);
-    }
-
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
-     * supplied serializers and deserializers, and the system time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keySerializer the serializer for keys; may not be null
-     * @param keyDeserializer the deserializer for keys; may not be null
-     * @param valueSerializer the serializer for values; may not be null
-     * @param valueDeserializer the deserializer for values; may not be null
-     * @return the key-value store
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context,
-                                                           Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
-                                                           Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) {
-        return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime());
-    }
-
-    /**
-     * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the
-     * supplied serializers and deserializers, and the given time provider.
-     * 
-     * @param name the name of the store, used in the name of the topic to which entries are recorded
-     * @param context the processing context
-     * @param keySerializer the serializer for keys; may not be null
-     * @param keyDeserializer the deserializer for keys; may not be null
-     * @param valueSerializer the serializer for values; may not be null
-     * @param valueDeserializer the deserializer for values; may not be null
-     * @param time the time provider; may be null if the system time should be used
-     * @return the key-value store
-     */
-    public static <K, V> RocksDBKeyValueStore<K, V> create(String name, ProcessorContext context,
-                                                           Serializer<K> keySerializer, Deserializer<K> keyDeserializer,
-                                                           Serializer<V> valueSerializer, Deserializer<V> valueDeserializer,
-                                                           Time time) {
-        Serdes<K, V> serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer);
-        return new RocksDBKeyValueStore<>(name, context, serdes, time);
-    }
-
     protected RocksDBKeyValueStore(String name, ProcessorContext context, Serdes<K, V> serdes, Time time) {
         super(name, new RocksDBStore<K, V>(name, context, serdes), context, serdes, "rocksdb-state", time != null ? time : new SystemTime());
     }
@@ -381,18 +265,18 @@ private static class RocksDBRangeIterator<K, V> extends RocksDbIterator<K, V> {
             // comparator to be pluggable, and the default is lexicographic, so it's
             // safe to just force lexicographic comparator here for now.
             private final Comparator<byte[]> comparator = new LexicographicComparator();
-            byte[] to;
+            byte[] rawToKey;
 
             public RocksDBRangeIterator(RocksIterator iter, Serdes<K, V> serdes,
                     K from, K to) {
                 super(iter, serdes);
                 iter.seek(serdes.rawKey(from));
-                this.to = serdes.rawKey(to);
+                this.rawToKey = serdes.rawKey(to);
             }
 
             @Override
             public boolean hasNext() {
-                return super.hasNext() && comparator.compare(super.peekRawKey(), this.to) < 0;
+                return super.hasNext() && comparator.compare(super.peekRawKey(), this.rawToKey) < 0;
             }
         }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
index 4fa21a78f397a..540d7634653f1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
@@ -39,7 +39,7 @@ public static <K, V> Serdes<K, V> withBuiltinTypes(String topic, Class<K> keyCla
     }
 
     @SuppressWarnings("unchecked")
-    private static <T> Serializer<T> serializer(Class<T> type) {
+    static <T> Serializer<T> serializer(Class<T> type) {
         if (String.class.isAssignableFrom(type)) return (Serializer<T>) new StringSerializer();
         if (Integer.class.isAssignableFrom(type)) return (Serializer<T>) new IntegerSerializer();
         if (Long.class.isAssignableFrom(type)) return (Serializer<T>) new LongSerializer();
@@ -48,7 +48,7 @@ private static <T> Serializer<T> serializer(Class<T> type) {
     }
 
     @SuppressWarnings("unchecked")
-    private static <T> Deserializer<T> deserializer(Class<T> type) {
+    static <T> Deserializer<T> deserializer(Class<T> type) {
         if (String.class.isAssignableFrom(type)) return (Deserializer<T>) new StringDeserializer();
         if (Integer.class.isAssignableFrom(type)) return (Deserializer<T>) new IntegerDeserializer();
         if (Long.class.isAssignableFrom(type)) return (Deserializer<T>) new LongDeserializer();
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
new file mode 100644
index 0000000000000..171ed44c76727
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
@@ -0,0 +1,257 @@
+/**
+ * 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.
+ */
+package org.apache.kafka.streams.state;
+
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+/**
+ * Factory for creating key-value stores.
+ */
+public class Stores {
+
+    /**
+     * Begin to create a new {@link org.apache.kafka.streams.processor.StateStore} instance.
+     * 
+     * @param name the name of the store
+     * @param context the processor context
+     * @return the factory that can be used to specify other options or configurations for the store; never null
+     */
+    public static StoreFactory create(final String name, final ProcessorContext context) {
+        return new StoreFactory() {
+            @Override
+            public <K> ValueFactory<K> withKeys(final Serializer<K> keySerializer, final Deserializer<K> keyDeserializer) {
+                return new ValueFactory<K>() {
+                    @Override
+                    public <V> KeyValueFactory<K, V> withValues(final Serializer<V> valueSerializer,
+                                                                final Deserializer<V> valueDeserializer) {
+                        final Serdes<K, V> serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer,
+                                context);
+                        return new KeyValueFactory<K, V>() {
+                            @Override
+                            public InMemoryKeyValueFactory<K, V> inMemory() {
+                                return new InMemoryKeyValueFactory<K, V>() {
+                                    @Override
+                                    public KeyValueStore<K, V> build() {
+                                        return new InMemoryKeyValueStore<>(name, context, serdes, null);
+                                    }
+                                };
+                            }
+
+                            @Override
+                            public LocalDatabaseKeyValueFactory<K, V> localDatabase() {
+                                return new LocalDatabaseKeyValueFactory<K, V>() {
+                                    @Override
+                                    public KeyValueStore<K, V> build() {
+                                        return new RocksDBKeyValueStore<>(name, context, serdes, null);
+                                    }
+                                };
+                            }
+                        };
+                    }
+                };
+            }
+        };
+    }
+
+    public static abstract class StoreFactory {
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the keys will be {@link String}s.
+         * 
+         * @return the interface used to specify the type of values; never null
+         */
+        public ValueFactory<String> withStringKeys() {
+            return withKeys(new StringSerializer(), new StringDeserializer());
+        }
+
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the keys will be {@link Integer}s.
+         * 
+         * @return the interface used to specify the type of values; never null
+         */
+        public ValueFactory<Integer> withIntegerKeys() {
+            return withKeys(new IntegerSerializer(), new IntegerDeserializer());
+        }
+
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the keys will be {@link Long}s.
+         * 
+         * @return the interface used to specify the type of values; never null
+         */
+        public ValueFactory<Long> withLongKeys() {
+            return withKeys(new LongSerializer(), new LongDeserializer());
+        }
+
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the keys will be byte arrays.
+         * 
+         * @return the interface used to specify the type of values; never null
+         */
+        public ValueFactory<byte[]> withByteArrayKeys() {
+            return withKeys(new ByteArraySerializer(), new ByteArrayDeserializer());
+        }
+
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the keys will be either {@link String}, {@link Integer},
+         * {@link Long}, or {@code byte[]}.
+         * 
+         * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and
+         *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+         *            {@code byte[].class})
+         * @return the interface used to specify the type of values; never null
+         */
+        public <K> ValueFactory<K> withKeys(Class<K> keyClass) {
+            return withKeys(Serdes.serializer(keyClass), Serdes.deserializer(keyClass));
+        }
+
+        /**
+         * Begin to create a {@link KeyValueStore} by specifying the serializer and deserializer for the keys.
+         * 
+         * @param keySerializer the serializer for keys; may not be null
+         * @param keyDeserializer the deserializer for keys; may not be null
+         * @return the interface used to specify the type of values; never null
+         */
+        public abstract <K> ValueFactory<K> withKeys(Serializer<K> keySerializer, Deserializer<K> keyDeserializer);
+    }
+
+    /**
+     * The interface used to specify the type of values for key-value stores.
+     * 
+     * @param <K> the type of keys
+     */
+    public static abstract class ValueFactory<K> {
+        /**
+         * Use {@link String} values.
+         * 
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public KeyValueFactory<K, String> withStringValues() {
+            return withValues(new StringSerializer(), new StringDeserializer());
+        }
+
+        /**
+         * Use {@link Integer} values.
+         * 
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public KeyValueFactory<K, Integer> withIntegerValues() {
+            return withValues(new IntegerSerializer(), new IntegerDeserializer());
+        }
+
+        /**
+         * Use {@link Long} values.
+         * 
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public KeyValueFactory<K, Long> withLongValues() {
+            return withValues(new LongSerializer(), new LongDeserializer());
+        }
+
+        /**
+         * Use byte arrays for values.
+         * 
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public KeyValueFactory<K, byte[]> withByteArrayValues() {
+            return withValues(new ByteArraySerializer(), new ByteArrayDeserializer());
+        }
+
+        /**
+         * Use values of the specified type, which must be either {@link String}, {@link Integer}, {@link Long}, or {@code byte[]}
+         * .
+         * 
+         * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and
+         *            deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or
+         *            {@code byte[].class})
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public <V> KeyValueFactory<K, V> withValues(Class<V> valueClass) {
+            return withValues(Serdes.serializer(valueClass), Serdes.deserializer(valueClass));
+        }
+
+        /**
+         * Use the specified serializer and deserializer for the values.
+         * 
+         * @param valueSerializer the serializer for value; may not be null
+         * @param valueDeserializer the deserializer for values; may not be null
+         * @return the interface used to specify the remaining key-value store options; never null
+         */
+        public abstract <V> KeyValueFactory<K, V> withValues(Serializer<V> valueSerializer, Deserializer<V> valueDeserializer);
+
+    }
+
+    /**
+     * The interface used to specify the different kinds of key-value stores.
+     *
+     * @param <K> the type of keys
+     * @param <V> the type of values
+     */
+    public static interface KeyValueFactory<K, V> {
+        /**
+         * Keep all key-value entries in-memory, although for durability all entries are recorded in a Kafka topic that can be
+         * read to restore the entries if they are lost.
+         * 
+         * @return the factory to create in-memory key-value stores; never null
+         */
+        InMemoryKeyValueFactory<K, V> inMemory();
+
+        /**
+         * Keep all key-value entries off-heap in a local database, although for durability all entries are recorded in a Kafka
+         * topic that can be read to restore the entries if they are lost.
+         * 
+         * @return the factory to create in-memory key-value stores; never null
+         */
+        LocalDatabaseKeyValueFactory<K, V> localDatabase();
+    }
+
+    /**
+     * The interface used to create in-memory key-value stores.
+     *
+     * @param <K> the type of keys
+     * @param <V> the type of values
+     */
+    public static interface InMemoryKeyValueFactory<K, V> {
+        /**
+         * Return the new key-value store.
+         * @return the key-value store; never null
+         */
+        KeyValueStore<K, V> build();
+    }
+
+    /**
+     * The interface used to create off-heap key-value stores that use a local database.
+     *
+     * @param <K> the type of keys
+     * @param <V> the type of values
+     */
+    public static interface LocalDatabaseKeyValueFactory<K, V> {
+        /**
+         * Return the new key-value store.
+         * @return the key-value store; never null
+         */
+        KeyValueStore<K, V> build();
+    }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index dd60ad4878cb0..5f8ca46cfd64f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -34,10 +34,10 @@
 import org.apache.kafka.streams.processor.ProcessorSupplier;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.apache.kafka.streams.processor.TopologyBuilder;
-import org.apache.kafka.streams.state.InMemoryKeyValueStore;
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StateUtils;
+import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.test.MockProcessorSupplier;
 import org.apache.kafka.test.ProcessorTopologyTestDriver;
 import org.junit.After;
@@ -264,7 +264,7 @@ public StatefulProcessor(String storeName) {
         @Override
         public void init(ProcessorContext context) {
             super.init(context);
-            store = InMemoryKeyValueStore.create(storeName, context, String.class, String.class);
+            store = Stores.create(storeName, context).withStringKeys().withStringValues().inMemory().build();
         }
 
         @Override
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
index 9a7b029412252..d8f06eaad226e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java
@@ -31,7 +31,7 @@ protected abstract <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorConte
                                                                       boolean useContextSerdes);
     
     @Test
-    public void testIntegerKeysAndStringValues() {
+    public void testPutGetRange() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create();
         KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, false);
@@ -97,7 +97,7 @@ else if (entry.key().equals(4))
     }
 
     @Test
-    public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializers() {
+    public void testPutGetRangeWithDefaultSerdes() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
         KeyValueStore<Integer, String> store = createKeyValueStore(driver.context(), Integer.class, String.class, true);
@@ -137,16 +137,16 @@ public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializer
     }
 
     @Test
-    public void testRestoringInetgerKeysAndValues() {
+    public void testRestore() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
 
         // Add any entries that will be restored to any store
         // that uses the driver's context ...
-        driver.addRestoreEntry(0, "zero");
-        driver.addRestoreEntry(1, "one");
-        driver.addRestoreEntry(2, "two");
-        driver.addRestoreEntry(4, "four");
+        driver.addEntryToRestoreLog(0, "zero");
+        driver.addEntryToRestoreLog(1, "one");
+        driver.addEntryToRestoreLog(2, "two");
+        driver.addEntryToRestoreLog(4, "four");
 
         // Create the store, which should register with the context and automatically
         // receive the restore entries ...
@@ -163,16 +163,16 @@ public void testRestoringInetgerKeysAndValues() {
     }
 
     @Test
-    public void testRestoringInetgerKeysAndValuesUsingDefaultSerializersAndDeserializers() {
+    public void testRestoreWithDefaultSerdes() {
         // Create the test driver ...
         KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
 
         // Add any entries that will be restored to any store
         // that uses the driver's context ...
-        driver.addRestoreEntry(0, "zero");
-        driver.addRestoreEntry(1, "one");
-        driver.addRestoreEntry(2, "two");
-        driver.addRestoreEntry(4, "four");
+        driver.addEntryToRestoreLog(0, "zero");
+        driver.addEntryToRestoreLog(1, "one");
+        driver.addEntryToRestoreLog(2, "two");
+        driver.addEntryToRestoreLog(4, "four");
 
         // Create the store, which should register with the context and automatically
         // receive the restore entries ...
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
index 9ca42b9f1d4a6..bee99676a56e1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java
@@ -16,16 +16,23 @@
  */
 package org.apache.kafka.streams.state;
 
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.processor.ProcessorContext;
 
 public class InMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest {
 
+    @SuppressWarnings("unchecked")
     @Override
     protected <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
                                                              boolean useContextSerdes) {
         if (useContextSerdes) {
-            return InMemoryKeyValueStore.create("my-store", context);
+            Serializer<K> keySer = (Serializer<K>) context.keySerializer();
+            Deserializer<K> keyDeser = (Deserializer<K>) context.keyDeserializer();
+            Serializer<V> valSer = (Serializer<V>) context.valueSerializer();
+            Deserializer<V> valDeser = (Deserializer<V>) context.valueDeserializer();
+            return Stores.create("my-store", context).withKeys(keySer, keyDeser).withValues(valSer, valDeser).inMemory().build();
         }
-        return InMemoryKeyValueStore.create("my-store", context, keyClass, valueClass);
+        return Stores.create("my-store", context).withKeys(keyClass).withValues(valueClass).inMemory().build();
     }
 }
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 25652eac20e43..8fdbfff3f6f8b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -49,12 +49,13 @@
  * This component can be used to help test a {@link KeyValueStore}'s ability to read and write entries.
  * 
  * <pre>
- * &#47;&#47; Create the test driver ...
+ * // Create the test driver ...
  * KeyValueStoreTestDriver&lt;Integer, String> driver = KeyValueStoreTestDriver.create();
- * InMemoryKeyValueStore&lt;Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
- *                                                                             Integer.class, String.class);
+ * KeyValueStore&lt;Integer, String> store = Stores.create("my-store", driver.context())
+ *                                              .withIntegerKeys().withStringKeys()
+ *                                              .inMemory().build();
  * 
- * &#47;&#47; Verify that the store reads and writes correctly ...
+ * // Verify that the store reads and writes correctly ...
  * store.put(0, "zero");
  * store.put(1, "one");
  * store.put(2, "two");
@@ -69,7 +70,7 @@
  * assertNull(store.get(3));
  * store.delete(5);
  * 
- * &#47;&#47; Flush the store and verify all current entries were properly flushed ...
+ * // Flush the store and verify all current entries were properly flushed ...
  * store.flush();
  * assertEquals("zero", driver.flushedEntryStored(0));
  * assertEquals("one", driver.flushedEntryStored(1));
@@ -87,30 +88,33 @@
  * <p>
  * <h2>Restoring a store</h2>
  * This component can be used to test whether a {@link KeyValueStore} implementation properly
- * {@link ProcessorContext#register(StateStore, RestoreFunc) registers itself} with the {@link ProcessorContext}. To do this,
- * create an instance of this driver component, {@link #addRestoreEntry(Object, Object) add entries} that will be passed to the
- * store upon creation, and then create the store using this driver's {@link #context() ProcessorContext}:
+ * {@link ProcessorContext#register(StateStore, StateRestoreCallback) registers itself} with the {@link ProcessorContext}, so that
+ * the persisted contents of a store are properly restored from the flushed entries when the store instance is started.
+ * <p>
+ * To do this, create an instance of this driver component, {@link #addEntryToRestoreLog(Object, Object) add entries} that will be
+ * passed to the store upon creation (simulating the entries that were previously flushed to the topic), and then create the store
+ * using this driver's {@link #context() ProcessorContext}:
  * 
  * <pre>
- * &#47;&#47; Create the test driver ...
+ * // Create the test driver ...
  * KeyValueStoreTestDriver&lt;Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
  * 
- * &#47;&#47; Add any entries that will be restored to any store
- * &#47;&#47; that uses the driver's context ...
+ * // Add any entries that will be restored to any store that uses the driver's context ...
  * driver.addRestoreEntry(0, "zero");
  * driver.addRestoreEntry(1, "one");
  * driver.addRestoreEntry(2, "two");
  * driver.addRestoreEntry(4, "four");
  * 
- * &#47;&#47; Create the store, which should register with the context and automatically
- * &#47;&#47; receive the restore entries ...
- * InMemoryKeyValueStore&lt;Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
- *                                                                             Integer.class, String.class);
+ * // Create the store, which should register with the context and automatically
+ * // receive the restore entries ...
+ * KeyValueStore&lt;Integer, String> store = Stores.create("my-store", driver.context())
+ *                                              .withIntegerKeys().withStringKeys()
+ *                                              .inMemory().build();
  * 
- * &#47;&#47; Verify that the store's contents were properly restored ...
+ * // Verify that the store's contents were properly restored ...
  * assertEquals(0, driver.checkForRestoredEntries(store));
  * 
- * &#47;&#47; and there are no other entries ...
+ * // and there are no other entries ...
  * assertEquals(4, driver.sizeOf(store));
  * </pre>
  * 
@@ -144,7 +148,7 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
             @Override
             public T deserialize(String topic, byte[] data) {
-                throw new UnsupportedOperationException("This serializer should not be used");
+                throw new UnsupportedOperationException("This deserializer should not be used");
             }
 
             @Override
@@ -220,6 +224,7 @@ public static <K, V> KeyValueStoreTestDriver<K, V> create(Serializer<K> keySeria
         public Sensor addLatencySensor(String scopeName, String entityName, String operationName, String... tags) {
             return null;
         }
+
         @Override
         public void recordLatency(Sensor sensor, long startNs, long endNs) {
         }
@@ -310,16 +315,35 @@ private void restoreEntries(StateRestoreCallback func) {
     }
 
     /**
-     * This method adds an entry to the "restore log" for the {@link KeyValueStore}. This should be called <em>before</em>
-     * creating
-     * the {@link KeyValueStore} with the {@link #context() ProcessorContext}; when the {@link KeyValueStore} is created, it will
-     * {@link ProcessorContext#register(StateStore, RestoreFunc) register} itself with the {@link #context() ProcessorContext},
-     * and this object will then pre-populate the store with all restore entries added via this method.
+     * This method adds an entry to the "restore log" for the {@link KeyValueStore}, and is used <em>only</em> when testing the
+     * restore functionality of a {@link KeyValueStore} implementation.
+     * <p>
+     * To create such a test, create the test driver, call this method one or more times, and then create the
+     * {@link KeyValueStore}. Your tests can then check whether the store contains the entries from the log.
+     * 
+     * <pre>
+     * // Set up the driver and pre-populate the log ...
+     * KeyValueStoreTestDriver&lt;Integer, String> driver = KeyValueStoreTestDriver.create();
+     * driver.addRestoreEntry(1,"value1");
+     * driver.addRestoreEntry(2,"value2");
+     * driver.addRestoreEntry(3,"value3");
+     * 
+     * // Create the store using the driver's context ...
+     * ProcessorContext context = driver.context();
+     * KeyValueStore&lt;Integer, String> store = ...
+     * 
+     * // Verify that the store's contents were properly restored from the log ...
+     * assertEquals(0, driver.checkForRestoredEntries(store));
+     * 
+     * // and there are no other entries ...
+     * assertEquals(3, driver.sizeOf(store));
+     * </pre>
      * 
      * @param key the key for the entry
      * @param value the value for the entry
+     * @see #checkForRestoredEntries(KeyValueStore)
      */
-    public void addRestoreEntry(K key, V value) {
+    public void addEntryToRestoreLog(K key, V value) {
         restorableEntries.add(new Entry<K, V>(key, value));
     }
 
@@ -328,11 +352,11 @@ public void addRestoreEntry(K key, V value) {
      * written by the store to the Kafka topic, making them available via the {@link #flushedEntryStored(Object)} and
      * {@link #flushedEntryRemoved(Object)} methods.
      * <p>
-     * If the {@link KeyValueStore}'s are to be restored upon its startup, be sure to {@link #addRestoreEntry(Object, Object)
+     * If the {@link KeyValueStore}'s are to be restored upon its startup, be sure to {@link #addEntryToRestoreLog(Object, Object)
      * add the restore entries} before creating the store with the {@link ProcessorContext} returned by this method.
      * 
      * @return the processing context; never null
-     * @see #addRestoreEntry(Object, Object)
+     * @see #addEntryToRestoreLog(Object, Object)
      */
     public ProcessorContext context() {
         return context;
@@ -349,11 +373,12 @@ public Iterable<Entry<K, V>> restoredEntries() {
     }
 
     /**
-     * Utility method that will count the number of {@link #addRestoreEntry(Object, Object) restore entries} missing from the
+     * Utility method that will count the number of {@link #addEntryToRestoreLog(Object, Object) restore entries} missing from the
      * supplied store.
      * 
      * @param store the store that is to have all of the {@link #restoredEntries() restore entries}
      * @return the number of restore entries missing from the store, or 0 if all restore entries were found
+     * @see #addEntryToRestoreLog(Object, Object)
      */
     public int checkForRestoredEntries(KeyValueStore<K, V> store) {
         int missing = 0;
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java
index d0c00fdb6477d..9ac17403cace8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java
@@ -16,16 +16,23 @@
  */
 package org.apache.kafka.streams.state;
 
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.processor.ProcessorContext;
 
 public class RocksDBKeyValueStoreTest extends AbstractKeyValueStoreTest {
 
+    @SuppressWarnings("unchecked")
     @Override
     protected <K, V> KeyValueStore<K, V> createKeyValueStore(ProcessorContext context, Class<K> keyClass, Class<V> valueClass,
                                                              boolean useContextSerdes) {
         if (useContextSerdes) {
-            return RocksDBKeyValueStore.create("my-store", context);
+            Serializer<K> keySer = (Serializer<K>) context.keySerializer();
+            Deserializer<K> keyDeser = (Deserializer<K>) context.keyDeserializer();
+            Serializer<V> valSer = (Serializer<V>) context.valueSerializer();
+            Deserializer<V> valDeser = (Deserializer<V>) context.valueDeserializer();
+            return Stores.create("my-store", context).withKeys(keySer, keyDeser).withValues(valSer, valDeser).localDatabase().build();
         }
-        return RocksDBKeyValueStore.create("my-store", context, keyClass, valueClass);
+        return Stores.create("my-store", context).withKeys(keyClass).withValues(valueClass).localDatabase().build();
     }
 }
\ No newline at end of file