Added an InMemoryLRUCacheStore and KeyValueStore unit test framework

streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java

@@ -48,7 +48,7 @@ public Processor<String, String> instance() {
                 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

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;
     }

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java

@@ -30,6 +30,10 @@
 import java.util.Map;
 
 public class RecordCollector {
+
+    public static interface Supplier {
+        public RecordCollector recordCollector();
+    }
 
     private static final Logger log = LoggerFactory.getLogger(RecordCollector.class);
 

streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java

@@ -18,6 +18,8 @@
 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;
 
@@ -35,26 +37,127 @@
  */
 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
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Class<K> keyClass, Class<V> valueClass) {
+        return new InMemoryKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass),
+                new SystemTime());
     }
 
-    public InMemoryKeyValueStore(String name, ProcessorContext context, Time time) {
-        super(name, new MemoryStore<K, V>(name, context), context, "kafka-streams", time);
+    /**
+     * 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 not be null
+     * @return the key-value store
+     */
+    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.
+     * 
+     * @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 not be null
+     * @return the key-value store
+     */
+    public static <K, V> InMemoryKeyValueStore<K, V> create(String name, ProcessorContext context, Time time) {
+        return new InMemoryKeyValueStore<>(name, context, new Serdes<K, V>(name, context), 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 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,
+                                                            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, "kafka-streams", time);
     }
 
     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
@@ -103,11 +206,6 @@ public void flush() {
             // do-nothing since it is in-memory
         }
 
-        public void restore() {
-            // this should not happen since it is in-memory, hence no state to load from disk
-            throw new IllegalStateException("This should not happen");
-        }
-
         @Override
         public void close() {
             // do-nothing