kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [1/2] kafka git commit: KAFKA-3452 Follow-up: Refactoring StateStore hierarchies
Date Tue, 17 Jan 2017 22:13:54 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 825f225bc -> 73b7ae001


http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplier.java
index 84f1734..abaaffd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplier.java
@@ -18,6 +18,7 @@
 package org.apache.kafka.streams.state.internals;
 
 import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.streams.state.WindowStore;
 
@@ -58,23 +59,35 @@ public class RocksDBWindowStoreSupplier<K, V> extends AbstractStoreSupplier<K, V
     }
 
     public WindowStore get() {
-        final RocksDBSegmentedBytesStore bytesStore = new RocksDBSegmentedBytesStore(name, retentionPeriod, numSegments, new WindowStoreKeySchema());
-        if (!enableCaching) {
-            final RocksDBWindowStore<K, V> segmentedStore = new RocksDBWindowStore<>(name, retainDuplicates, keySerde, valueSerde,
-                                                                                     logged ? new ChangeLoggingSegmentedBytesStore(bytesStore)
-                                                                                                   : bytesStore);
-            return new MeteredWindowStore<>(segmentedStore, "rocksdb-window", time);
-        }
+        return maybeWrapCaching(
+                maybeWrapLogged(
+                        new RocksDBSegmentedBytesStore(
+                                name,
+                                retentionPeriod,
+                                numSegments,
+                                new WindowStoreKeySchema()
+                        )));
 
-        return new CachingWindowStore<>(new MeteredSegmentedBytesStore(logged ? new ChangeLoggingSegmentedBytesStore(bytesStore)
-                                                                               : bytesStore,
-                                                                       "rocksdb-window",
-                                                                       time),
-                                        keySerde, valueSerde, windowSize);
     }
 
     @Override
     public long retentionPeriod() {
         return retentionPeriod;
     }
+
+    private SegmentedBytesStore maybeWrapLogged(final SegmentedBytesStore inner) {
+        if (!logged) {
+            return inner;
+        }
+        return new ChangeLoggingSegmentedBytesStore(inner);
+    }
+
+    private WindowStore<K, V> maybeWrapCaching(final SegmentedBytesStore inner) {
+        final MeteredSegmentedBytesStore metered = new MeteredSegmentedBytesStore(inner, "rocksdb-window", time);
+        if (!enableCaching) {
+            return new RocksDBWindowStore<>(metered, keySerde, valueSerde, retainDuplicates);
+        }
+        final RocksDBWindowStore<Bytes, byte[]> windowed = RocksDBWindowStore.bytesStore(metered, retainDuplicates);
+        return new CachingWindowStore<>(windowed, keySerde, valueSerde, windowSize);
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/main/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIterator.java
new file mode 100644
index 0000000..d76e8a4
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIterator.java
@@ -0,0 +1,70 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.StateSerdes;
+
+import java.util.NoSuchElementException;
+
+class SerializedKeyValueIterator<K, V> implements KeyValueIterator<K, V> {
+
+    private final KeyValueIterator<Bytes, byte[]> bytesIterator;
+    private final StateSerdes<K, V> serdes;
+
+    SerializedKeyValueIterator(final KeyValueIterator<Bytes, byte[]> bytesIterator,
+                               final StateSerdes<K, V> serdes) {
+
+        this.bytesIterator = bytesIterator;
+        this.serdes = serdes;
+    }
+
+    @Override
+    public void close() {
+        bytesIterator.close();
+    }
+
+    @Override
+    public K peekNextKey() {
+        if (!hasNext()) {
+            throw new NoSuchElementException();
+        }
+        final Bytes bytes = bytesIterator.peekNextKey();
+        return serdes.keyFrom(bytes.get());
+    }
+
+    @Override
+    public boolean hasNext() {
+        return bytesIterator.hasNext();
+    }
+
+    @Override
+    public KeyValue<K, V> next() {
+        if (!hasNext()) {
+            throw new NoSuchElementException();
+        }
+        final KeyValue<Bytes, byte[]> next = bytesIterator.next();
+        return KeyValue.pair(serdes.keyFrom(next.key.get()), serdes.valueFrom(next.value));
+    }
+
+    @Override
+    public void remove() {
+        throw new UnsupportedOperationException("remove not supported by SerializedKeyValueIterator");
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreUtils.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreUtils.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreUtils.java
index 1ea6bef..074cf8a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreUtils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreUtils.java
@@ -39,7 +39,10 @@ public class WindowStoreUtils {
 
     public static <K> byte[] toBinaryKey(K key, final long timestamp, final int seqnum, StateSerdes<K, ?> serdes) {
         byte[] serializedKey = serdes.rawKey(key);
+        return toBinaryKey(serializedKey, timestamp, seqnum);
+    }
 
+    static byte[] toBinaryKey(byte[] serializedKey, final long timestamp, final int seqnum) {
         ByteBuffer buf = ByteBuffer.allocate(serializedKey.length + TIMESTAMP_SIZE + SEQNUM_SIZE);
         buf.put(serializedKey);
         buf.putLong(timestamp);

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
new file mode 100644
index 0000000..3d80b98
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
@@ -0,0 +1,90 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.streams.errors.InvalidStateStoreException;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStore;
+
+/**
+ * A storage engine wrapper for utilities like logging, caching, and metering.
+ */
+interface WrappedStateStore extends StateStore {
+
+    /**
+     * Return the inner storage engine
+     *
+     * @return wrapped inner storage engine
+     */
+    StateStore inner();
+
+    abstract class AbstractWrappedStateStore implements WrappedStateStore {
+        final StateStore innerState;
+
+        AbstractWrappedStateStore(StateStore inner) {
+            this.innerState = inner;
+        }
+
+        @Override
+        public void init(ProcessorContext context, StateStore root) {
+            innerState.init(context, root);
+        }
+
+        @Override
+        public String name() {
+            return innerState.name();
+        }
+
+        @Override
+        public boolean persistent() {
+            return innerState.persistent();
+        }
+
+        @Override
+        public boolean isOpen() {
+            return innerState.isOpen();
+        }
+
+        void validateStoreOpen() {
+            if (!innerState.isOpen()) {
+                throw new InvalidStateStoreException("Store " + innerState.name() + " is currently closed.");
+            }
+        }
+
+        @Override
+        public StateStore inner() {
+            if (innerState instanceof WrappedStateStore) {
+                return ((WrappedStateStore) innerState).inner();
+            }
+            return innerState;
+        }
+
+        @Override
+        public void flush() {
+            innerState.flush();
+        }
+
+        @Override
+        public void close() {
+            innerState.close();
+        }
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
index 729e190..b6d8a97 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
@@ -18,6 +18,7 @@
 package org.apache.kafka.streams.kstream.internals;
 
 import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.Aggregator;
 import org.apache.kafka.streams.kstream.ForeachAction;
 import org.apache.kafka.streams.kstream.Initializer;
@@ -40,9 +41,14 @@ import org.apache.kafka.test.TestUtils;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 
 public class KGroupedStreamImplTest {
@@ -337,4 +343,38 @@ public class KGroupedStreamImplTest {
     public void shouldNotAcceptNullStoreStoreSupplierNameWhenCountingSessionWindows() throws Exception {
         groupedStream.count(SessionWindows.with(90), (StateStoreSupplier<SessionStore>) null);
     }
-}
+
+    @Test
+    public void shouldCountWindowed() throws Exception {
+        final List<KeyValue<Windowed<String>, Long>> results = new ArrayList<>();
+        groupedStream.count(
+                TimeWindows.of(500L),
+                "aggregate-by-key-windowed")
+                .foreach(new ForeachAction<Windowed<String>, Long>() {
+                    @Override
+                    public void apply(final Windowed<String> key, final Long value) {
+                        results.add(KeyValue.pair(key, value));
+                    }
+                });
+
+        final KStreamTestDriver driver = new KStreamTestDriver(builder, TestUtils.tempDirectory(), 0);
+        driver.setTime(0);
+        driver.process(TOPIC, "1", "A");
+        driver.process(TOPIC, "2", "B");
+        driver.process(TOPIC, "3", "C");
+        driver.setTime(500);
+        driver.process(TOPIC, "1", "A");
+        driver.process(TOPIC, "1", "A");
+        driver.process(TOPIC, "2", "B");
+        driver.process(TOPIC, "2", "B");
+        assertThat(results, equalTo(Arrays.asList(
+                KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), 1L),
+                KeyValue.pair(new Windowed<>("2", new TimeWindow(0, 500)), 1L),
+                KeyValue.pair(new Windowed<>("3", new TimeWindow(0, 500)), 1L),
+                KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), 1L),
+                KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), 2L),
+                KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), 1L),
+                KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), 2L)
+        )));
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
index c603aa0..5035f70 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
@@ -59,7 +59,8 @@ public class CachingSessionStoreTest {
     @Before
     public void setUp() throws Exception {
         underlying = new RocksDBSegmentedBytesStore("test", 60000, 3, new SessionKeySchema());
-        cachingStore = new CachingSessionStore<>(underlying,
+        final RocksDBSessionStore<Bytes, byte[]> sessionStore = new RocksDBSessionStore<>(underlying, Serdes.Bytes(), Serdes.ByteArray());
+        cachingStore = new CachingSessionStore<>(sessionStore,
                                                  Serdes.String(),
                                                  Serdes.Long());
         cache = new ThreadCache("testCache", MAX_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
index 37fc9a0..1de1002 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingWindowStoreTest.java
@@ -54,13 +54,15 @@ public class CachingWindowStoreTest {
     private String topic;
     private static final long DEFAULT_TIMESTAMP = 10L;
     private WindowStoreKeySchema keySchema;
+    private RocksDBWindowStore<Bytes, byte[]> windowStore;
 
     @Before
     public void setUp() throws Exception {
         keySchema = new WindowStoreKeySchema();
         underlying = new RocksDBSegmentedBytesStore("test", 30000, 3, keySchema);
+        windowStore = new RocksDBWindowStore<>(underlying, Serdes.Bytes(), Serdes.ByteArray(), false);
         cacheListener = new CachingKeyValueStoreTest.CacheFlushListenerStub<>();
-        cachingStore = new CachingWindowStore<>(underlying,
+        cachingStore = new CachingWindowStore<>(windowStore,
                                                 Serdes.String(),
                                                 Serdes.String(),
                                                 WINDOW_SIZE);
@@ -72,6 +74,7 @@ public class CachingWindowStoreTest {
         cachingStore.init(context, cachingStore);
     }
 
+
     @Test
     public void shouldPutFetchFromCache() throws Exception {
         cachingStore.put("a", "a");

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
new file mode 100644
index 0000000..82fb831
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
@@ -0,0 +1,165 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.test.InMemoryKeyValueStore;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.NoOpRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class ChangeLoggingKeyValueBytesStoreTest {
+
+    private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv");
+    private final ChangeLoggingKeyValueBytesStore store = new ChangeLoggingKeyValueBytesStore(inner);
+    private final Map sent = new HashMap<>();
+    private final Bytes hi = Bytes.wrap("hi".getBytes());
+    private final Bytes hello = Bytes.wrap("hello".getBytes());
+    private final byte[] there = "there".getBytes();
+    private final byte[] world = "world".getBytes();
+
+    @Before
+    public void before() {
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                sent.put(record.key(), record.value());
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.Long(),
+                                                                      collector,
+                                                                      new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
+        context.setTime(0);
+        store.init(context, store);
+    }
+
+    @Test
+    public void shouldWriteKeyValueBytesToInnerStoreOnPut() throws Exception {
+        store.put(hi, there);
+        assertThat(inner.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldLogChangeOnPut() throws Exception {
+        store.put(hi, there);
+        assertThat((byte[]) sent.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldWriteAllKeyValueToInnerStoreOnPutAll() throws Exception {
+        store.putAll(Arrays.asList(KeyValue.pair(hi, there),
+                                   KeyValue.pair(hello, world)));
+        assertThat(inner.get(hi), equalTo(there));
+        assertThat(inner.get(hello), equalTo(world));
+    }
+
+    @Test
+    public void shouldLogChangesOnPutAll() throws Exception {
+        store.putAll(Arrays.asList(KeyValue.pair(hi, there),
+                                   KeyValue.pair(hello, world)));
+        assertThat((byte[]) sent.get(hi), equalTo(there));
+        assertThat((byte[]) sent.get(hello), equalTo(world));
+    }
+
+    @Test
+    public void shouldPutNullOnDelete() throws Exception {
+        store.put(hi, there);
+        store.delete(hi);
+        assertThat(inner.get(hi), nullValue());
+    }
+
+    @Test
+    public void shouldReturnOldValueOnDelete() throws Exception {
+        store.put(hi, there);
+        assertThat(store.delete(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldLogKeyNullOnDelete() throws Exception {
+        store.put(hi, there);
+        store.delete(hi);
+        assertThat(sent.get(hi), nullValue());
+    }
+
+    @Test
+    public void shouldWriteToInnerOnPutIfAbsentNoPreviousValue() throws Exception {
+        store.putIfAbsent(hi, there);
+        assertThat(inner.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldNotWriteToInnerOnPutIfAbsentWhenValueForKeyExists() throws Exception {
+        store.put(hi, there);
+        store.putIfAbsent(hi, world);
+        assertThat(inner.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldWriteToChangelogOnPutIfAbsentWhenNoPreviousValue() throws Exception {
+        store.putIfAbsent(hi, there);
+        assertThat((byte[]) sent.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldNotWriteToChangeLogOnPutIfAbsentWhenValueForKeyExists() throws Exception {
+        store.put(hi, there);
+        store.putIfAbsent(hi, world);
+        assertThat((byte[]) sent.get(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldReturnCurrentValueOnPutIfAbsent() throws Exception {
+        store.put(hi, there);
+        assertThat(store.putIfAbsent(hi, world), equalTo(there));
+    }
+
+    @Test
+    public void shouldReturnNullOnPutIfAbsentWhenNoPreviousValue() throws Exception {
+        assertThat(store.putIfAbsent(hi, there), is(nullValue()));
+    }
+
+    @Test
+    public void shouldReturnValueOnGetWhenExists() throws Exception {
+        store.put(hello, world);
+        assertThat(store.get(hello), equalTo(world));
+    }
+
+    @Test
+    public void shouldReturnNullOnGetWhenDoesntExist() throws Exception {
+        assertThat(store.get(hello), is(nullValue()));
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java
new file mode 100644
index 0000000..8815c5a
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java
@@ -0,0 +1,207 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.test.InMemoryKeyValueStore;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.NoOpRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+
+public class ChangeLoggingKeyValueStoreTest {
+
+    private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv");
+    private final Serde<String> keySerde = Serdes.String();
+    private final Serde<String> valueSerde = Serdes.String();
+    private final ChangeLoggingKeyValueStore<String, String> store
+            = new ChangeLoggingKeyValueStore<>(inner, keySerde, valueSerde);
+    private final Map sent = new HashMap<>();
+    private final String hi = "hi";
+    private final Bytes hiBytes = Bytes.wrap(hi.getBytes());
+    private final String there = "there";
+    private final byte[] thereBytes = "there".getBytes();
+    private final String hello = "hello";
+    private final String world = "world";
+
+    @Before
+    public void before() {
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                sent.put(record.key(), record.value());
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.Long(),
+                                                                      collector,
+                                                                      new ThreadCache("testCache", 0, new MockStreamsMetrics(new Metrics())));
+        context.setTime(0);
+        store.init(context, store);
+    }
+
+    @Test
+    public void shouldWriteKeyValueBytesToInnerStoreOnPut() throws Exception {
+        store.put(hi, there);
+        assertThat(deserializedValueFromInner(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldLogChangeOnPut() throws Exception {
+        store.put(hi, there);
+        assertThat((byte[]) sent.get(hiBytes), equalTo(thereBytes));
+    }
+
+    @Test
+    public void shouldWriteAllKeyValueToInnerStoreOnPutAll() throws Exception {
+        store.putAll(Arrays.asList(KeyValue.pair(hello, world),
+                                   KeyValue.pair(hi, there)));
+        assertThat(deserializedValueFromInner(hello), equalTo(world));
+        assertThat(deserializedValueFromInner(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldLogChangesOnPutAll() throws Exception {
+        store.putAll(Arrays.asList(KeyValue.pair(hi, there),
+                                   KeyValue.pair(hello, world)));
+        assertThat((byte[]) sent.get(hiBytes), equalTo(thereBytes));
+        assertThat((byte[]) sent.get(Bytes.wrap(hello.getBytes())), equalTo(world.getBytes()));
+    }
+
+    @Test
+    public void shouldPutNullOnDelete() throws Exception {
+        store.put(hi, there);
+        store.delete(hi);
+        assertThat(inner.get(hiBytes), nullValue());
+    }
+
+    @Test
+    public void shouldReturnOldValueOnDelete() throws Exception {
+        store.put(hi, there);
+        assertThat(store.delete(hi), equalTo(there));
+    }
+
+    @Test
+    public void shouldReturnNullOnDeleteIfNoOldValue() throws Exception {
+        assertThat(store.delete(hi), is(nullValue()));
+    }
+
+    @Test
+    public void shouldLogKeyNullOnDelete() throws Exception {
+        store.put(hi, there);
+        store.delete(hi);
+        assertThat(sent.get(hi), nullValue());
+    }
+
+    @Test
+    public void shouldWriteToInnerOnPutIfAbsentNoPreviousValue() throws Exception {
+        store.putIfAbsent(hi, there);
+        assertThat(inner.get(hiBytes), equalTo(thereBytes));
+    }
+
+    @Test
+    public void shouldNotWriteToInnerOnPutIfAbsentWhenValueForKeyExists() throws Exception {
+        store.put(hi, there);
+        store.putIfAbsent(hi, world);
+        assertThat(inner.get(hiBytes), equalTo(thereBytes));
+    }
+
+    @Test
+    public void shouldWriteToChangelogOnPutIfAbsentWhenNoPreviousValue() throws Exception {
+        store.putIfAbsent(hi, there);
+        assertThat((byte[]) sent.get(hiBytes), equalTo(thereBytes));
+    }
+
+    @Test
+    public void shouldNotWriteToChangeLogOnPutIfAbsentWhenValueForKeyExists() throws Exception {
+        store.put(hi, there);
+        store.putIfAbsent(hi, world);
+        assertThat((byte[]) sent.get(hiBytes), equalTo(thereBytes));
+    }
+
+    @Test
+    public void shouldReturnCurrentValueOnPutIfAbsent() throws Exception {
+        store.put(hi, there);
+        assertThat(store.putIfAbsent(hi, world), equalTo(there));
+    }
+
+    @Test
+    public void shouldReturnNullOnPutIfAbsentWhenNoPreviousValue() throws Exception {
+        assertThat(store.putIfAbsent(hi, there), is(nullValue()));
+    }
+
+    @Test
+    public void shouldQueryRange() throws Exception {
+        store.put(hello, world);
+        store.put(hi, there);
+        store.put("zooom", "home");
+        final KeyValueIterator<String, String> range = store.range(hello, hi);
+        assertThat(range.next(), equalTo(KeyValue.pair(hello, world)));
+        assertThat(range.next(), equalTo(KeyValue.pair(hi, there)));
+        assertFalse(range.hasNext());
+    }
+
+    @Test
+    public void shouldReturnAllKeyValues() throws Exception {
+        store.put(hello, world);
+        store.put(hi, there);
+        final String zooom = "zooom";
+        final String home = "home";
+        store.put(zooom, home);
+        final KeyValueIterator<String, String> all = store.all();
+        assertThat(all.next(), equalTo(KeyValue.pair(hello, world)));
+        assertThat(all.next(), equalTo(KeyValue.pair(hi, there)));
+        assertThat(all.next(), equalTo(KeyValue.pair(zooom, home)));
+        assertFalse(all.hasNext());
+    }
+
+    @Test
+    public void shouldReturnValueOnGetWhenExists() throws Exception {
+        store.put(hello, world);
+        assertThat(store.get(hello), equalTo(world));
+    }
+
+    @Test
+    public void shouldReturnNullOnGetWhenDoesntExist() throws Exception {
+        assertThat(store.get(hello), is(nullValue()));
+    }
+
+    private String deserializedValueFromInner(final String key) {
+        return valueSerde.deserializer().deserialize("blah", inner.get(Bytes.wrap(key.getBytes())));
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIteratorTest.java
index 50845e8..0ebdd5c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIteratorTest.java
@@ -17,6 +17,7 @@
 
 package org.apache.kafka.streams.state.internals;
 
+import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.test.InMemoryKeyValueStore;
 import org.junit.Before;
 import org.junit.Test;
@@ -37,7 +38,7 @@ public class DelegatingPeekingKeyValueIteratorTest {
     }
 
     @Test
-    public void shouldPeekNext() throws Exception {
+    public void shouldPeekNextKey() throws Exception {
         store.put("A", "A");
         final DelegatingPeekingKeyValueIterator<String, String> peekingIterator = new DelegatingPeekingKeyValueIterator<>(name, store.all());
         assertEquals("A", peekingIterator.peekNextKey());
@@ -46,6 +47,15 @@ public class DelegatingPeekingKeyValueIteratorTest {
     }
 
     @Test
+    public void shouldPeekNext() throws Exception {
+        store.put("A", "A");
+        final DelegatingPeekingKeyValueIterator<String, String> peekingIterator = new DelegatingPeekingKeyValueIterator<>(name, store.all());
+        assertEquals(KeyValue.pair("A", "A"), peekingIterator.peekNext());
+        assertEquals(KeyValue.pair("A", "A"), peekingIterator.peekNext());
+        assertTrue(peekingIterator.hasNext());
+    }
+
+    @Test
     public void shouldPeekAndIterate() throws Exception {
         final String[] kvs = {"a", "b", "c", "d", "e", "f"};
         for (String kv : kvs) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIteratorTest.java
new file mode 100644
index 0000000..e7c2eb3
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIteratorTest.java
@@ -0,0 +1,113 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.streams.state.StateSerdes;
+import org.apache.kafka.test.KeyValueIteratorStub;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Iterator;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class MergedSortedCacheSessionStoreIteratorTest {
+
+    private final String storeKey = "a";
+    private final String cacheKey = "b";
+
+    private final SessionWindow storeWindow = new SessionWindow(0, 1);
+    private final Iterator<KeyValue<Windowed<Bytes>, byte[]>> storeKvs = Collections.singleton(
+            KeyValue.pair(new Windowed<>(Bytes.wrap(storeKey.getBytes()), storeWindow), storeKey.getBytes())).iterator();
+    private final SessionWindow cacheWindow = new SessionWindow(10, 20);
+    private final Iterator<KeyValue<Bytes, LRUCacheEntry>> cacheKvs = Collections.singleton(KeyValue.pair(
+            SessionKeySerde.toBinary(
+                    new Windowed<>(cacheKey, cacheWindow), Serdes.String().serializer()), new LRUCacheEntry(cacheKey.getBytes())))
+            .iterator();
+
+    @Test
+    public void shouldHaveNextFromStore() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(storeKvs, Collections.<KeyValue<Bytes, LRUCacheEntry>>emptyIterator());
+        assertTrue(mergeIterator.hasNext());
+    }
+
+    @Test
+    public void shouldGetNextFromStore() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(storeKvs, Collections.<KeyValue<Bytes, LRUCacheEntry>>emptyIterator());
+        assertThat(mergeIterator.next(), equalTo(KeyValue.pair(new Windowed<>(storeKey, storeWindow), storeKey)));
+    }
+
+    @Test
+    public void shouldPeekNextKeyFromStore() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(storeKvs, Collections.<KeyValue<Bytes, LRUCacheEntry>>emptyIterator());
+        assertThat(mergeIterator.peekNextKey(), equalTo(new Windowed<>(storeKey, storeWindow)));
+    }
+
+    @Test
+    public void shouldHaveNextFromCache() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(Collections.<KeyValue<Windowed<Bytes>, byte[]>>emptyIterator(),
+                                 cacheKvs);
+        assertTrue(mergeIterator.hasNext());
+    }
+
+    @Test
+    public void shouldGetNextFromCache() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(Collections.<KeyValue<Windowed<Bytes>, byte[]>>emptyIterator(), cacheKvs);
+        assertThat(mergeIterator.next(), equalTo(KeyValue.pair(new Windowed<>(cacheKey, cacheWindow), cacheKey)));
+    }
+
+    @Test
+    public void shouldPeekNextKeyFromCache() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> mergeIterator
+                = createIterator(Collections.<KeyValue<Windowed<Bytes>, byte[]>>emptyIterator(), cacheKvs);
+        assertThat(mergeIterator.peekNextKey(), equalTo(new Windowed<>(cacheKey, cacheWindow)));
+    }
+
+    @Test
+    public void shouldIterateBothStoreAndCache() throws Exception {
+        final MergedSortedCacheSessionStoreIterator<String, String> iterator = createIterator(storeKvs, cacheKvs);
+        assertThat(iterator.next(), equalTo(KeyValue.pair(new Windowed<>(storeKey, storeWindow), storeKey)));
+        assertThat(iterator.next(), equalTo(KeyValue.pair(new Windowed<>(cacheKey, cacheWindow), cacheKey)));
+        assertFalse(iterator.hasNext());
+    }
+
+    private MergedSortedCacheSessionStoreIterator<String, String> createIterator(final Iterator<KeyValue<Windowed<Bytes>, byte[]>> storeKvs,
+                                                                                 final Iterator<KeyValue<Bytes, LRUCacheEntry>> cacheKvs) {
+        final DelegatingPeekingKeyValueIterator<Windowed<Bytes>, byte[]> storeIterator
+                = new DelegatingPeekingKeyValueIterator<>("store", new KeyValueIteratorStub<>(storeKvs));
+
+        final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator
+                = new DelegatingPeekingKeyValueIterator<>("cache", new KeyValueIteratorStub<>(cacheKvs));
+        return new MergedSortedCacheSessionStoreIterator<>(cacheIterator, storeIterator, new StateSerdes<>("name", Serdes.String(), Serdes.String()));
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
index b04f248..376fca8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
@@ -19,7 +19,6 @@ package org.apache.kafka.streams.state.internals;
 
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
 import org.apache.kafka.streams.state.KeyValueIterator;
@@ -30,23 +29,25 @@ import org.junit.Test;
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 
 public class MergedSortedCacheWindowStoreIteratorTest {
 
+    private final List<KeyValue<Long, byte[]>> windowStoreKvPairs = new ArrayList<>();
+    private final ThreadCache cache = new ThreadCache("testCache", 1000000L,  new MockStreamsMetrics(new Metrics()));
+    private final String namespace = "one";
+    private final StateSerdes<String, String> stateSerdes = new StateSerdes<>("foo", Serdes.String(), Serdes.String());
+
     @Test
     public void shouldIterateOverValueFromBothIterators() throws Exception {
-        final List<KeyValue<Bytes, byte[]>> storeValues = new ArrayList<>();
-        final ThreadCache cache = new ThreadCache("testCache", 1000000L, new MockStreamsMetrics(new Metrics()));
-        final String namespace = "one";
-        final StateSerdes<String, String> stateSerdes = new StateSerdes<>("foo", Serdes.String(), Serdes.String());
         final List<KeyValue<Long, byte[]>> expectedKvPairs = new ArrayList<>();
-
         for (long t = 0; t < 100; t += 20) {
             final byte[] v1Bytes = String.valueOf(t).getBytes();
-            final KeyValue<Bytes, byte[]> v1 = KeyValue.pair(Bytes.wrap(WindowStoreUtils.toBinaryKey("a", t, 0, stateSerdes)), v1Bytes);
-            storeValues.add(v1);
+            final KeyValue<Long, byte[]> v1 = KeyValue.pair(t, v1Bytes);
+            windowStoreKvPairs.add(v1);
             expectedKvPairs.add(KeyValue.pair(t, v1Bytes));
             final byte[] keyBytes = WindowStoreUtils.toBinaryKey("a", t + 10, 0, stateSerdes);
             final byte[] valBytes = String.valueOf(t + 10).getBytes();
@@ -56,11 +57,11 @@ public class MergedSortedCacheWindowStoreIteratorTest {
 
         byte[] binaryFrom = WindowStoreUtils.toBinaryKey("a", 0, 0, stateSerdes);
         byte[] binaryTo = WindowStoreUtils.toBinaryKey("a", 100, 0, stateSerdes);
-        final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("name", new KeyValueIteratorStub<>(storeValues.iterator()));
+        final KeyValueIterator<Long, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("name", new KeyValueIteratorStub<>(windowStoreKvPairs.iterator()));
 
         final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, binaryFrom, binaryTo);
 
-        final MergedSortedCachedWindowStoreIterator<Bytes, byte[]> iterator = new MergedSortedCachedWindowStoreIterator<>(cacheIterator, storeIterator, new StateSerdes<>("name", Serdes.Bytes(), Serdes.ByteArray()));
+        final MergedSortedCacheWindowStoreIterator<byte[]> iterator = new MergedSortedCacheWindowStoreIterator<>(cacheIterator, storeIterator, new StateSerdes<>("name", Serdes.Long(), Serdes.ByteArray()));
         int index = 0;
         while (iterator.hasNext()) {
             final KeyValue<Long, byte[]> next = iterator.next();
@@ -70,4 +71,18 @@ public class MergedSortedCacheWindowStoreIteratorTest {
         }
     }
 
+    @Test
+    public void shouldPeekNextKey() throws Exception {
+        windowStoreKvPairs.add(KeyValue.pair(10L, "a".getBytes()));
+        cache.put(namespace, WindowStoreUtils.toBinaryKey("a", 0, 0, stateSerdes), new LRUCacheEntry("b".getBytes()));
+        byte[] binaryFrom = WindowStoreUtils.toBinaryKey("a", 0, 0, stateSerdes);
+        byte[] binaryTo = WindowStoreUtils.toBinaryKey("a", 100, 0, stateSerdes);
+        final KeyValueIterator<Long, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("name", new KeyValueIteratorStub<>(windowStoreKvPairs.iterator()));
+        final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, binaryFrom, binaryTo);
+        final MergedSortedCacheWindowStoreIterator<byte[]> iterator = new MergedSortedCacheWindowStoreIterator<>(cacheIterator, storeIterator, new StateSerdes<>("name", Serdes.Long(), Serdes.ByteArray()));
+        assertThat(iterator.peekNextKey(), equalTo(0L));
+        iterator.next();
+        assertThat(iterator.peekNextKey(), equalTo(10L));
+    }
+
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/ReadOnlyWindowStoreStub.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ReadOnlyWindowStoreStub.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ReadOnlyWindowStoreStub.java
index 2082e00..a2ce96c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ReadOnlyWindowStoreStub.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ReadOnlyWindowStoreStub.java
@@ -105,6 +105,11 @@ public class ReadOnlyWindowStoreStub<K, V> implements ReadOnlyWindowStore<K, V>,
         }
 
         @Override
+        public Long peekNextKey() {
+            throw new UnsupportedOperationException("peekNextKey not supported in stub");
+        }
+
+        @Override
         public boolean hasNext() {
             return underlying.hasNext();
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java
new file mode 100644
index 0000000..3d9a56c
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java
@@ -0,0 +1,155 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.NoOpRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class RocksDBKeyValueStoreSupplierTest {
+
+    private static final String STORE_NAME = "name";
+    private final ThreadCache cache = new ThreadCache("test", 1024, new MockStreamsMetrics(new Metrics()));
+    private final MockProcessorContext context = new MockProcessorContext(null,
+                                                                          TestUtils.tempDirectory(),
+                                                                          Serdes.String(),
+                                                                          Serdes.String(),
+                                                                          new NoOpRecordCollector(),
+                                                                          cache);
+    private KeyValueStore<String, String> store;
+
+    @After
+    public void close() {
+        store.close();
+    }
+
+    @Test
+    public void shouldCreateLoggingEnabledStoreWhenStoreLogged() throws Exception {
+        store = createStore(true, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put("a", "b");
+        assertFalse(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldNotBeLoggingEnabledStoreWhenLoggingNotEnabled() throws Exception {
+        store = createStore(false, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put("a", "b");
+        assertTrue(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldReturnCachedKeyValueStoreWhenCachingEnabled() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        context.setTime(1);
+        store.put("a", "b");
+        store.put("b", "c");
+        assertThat(store, is(instanceOf(CachingKeyValueStore.class)));
+        assertThat(cache.size(), is(2L));
+    }
+
+    @Test
+    public void shouldReturnMeteredStoreWhenCachingAndLoggingDisabled() throws Exception {
+        store = createStore(false, false);
+        assertThat(store, is(instanceOf(MeteredKeyValueStore.class)));
+    }
+
+    @Test
+    public void shouldReturnMeteredStoreWhenCachingDisabled() throws Exception {
+        store = createStore(true, false);
+        assertThat(store, is(instanceOf(MeteredKeyValueStore.class)));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenCached() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenLogged() throws Exception {
+        store = createStore(true, false);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    private KeyValueStore<String, String> createStore(final boolean logged, final boolean cached) {
+        return new RocksDBKeyValueStoreSupplier<>(STORE_NAME,
+                                                  Serdes.String(),
+                                                  Serdes.String(),
+                                                  logged,
+                                                  Collections.EMPTY_MAP,
+                                                  cached).get();
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java
new file mode 100644
index 0000000..28196a2
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java
@@ -0,0 +1,169 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.NoOpRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class RocksDBSessionStoreSupplierTest {
+
+    private static final String STORE_NAME = "name";
+    private final ThreadCache cache = new ThreadCache("test", 1024, new MockStreamsMetrics(new Metrics()));
+    private final MockProcessorContext context = new MockProcessorContext(null,
+                                                                          TestUtils.tempDirectory(),
+                                                                          Serdes.String(),
+                                                                          Serdes.String(),
+                                                                          new NoOpRecordCollector(),
+                                                                          cache);
+
+    private SessionStore<String, String> store;
+
+    @After
+    public void close() {
+        store.close();
+    }
+
+    @Test
+    public void shouldCreateLoggingEnabledStoreWhenStoreLogged() throws Exception {
+        store = createStore(true, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put(new Windowed<>("a", new SessionWindow(0, 10)), "b");
+        assertFalse(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldNotBeLoggingEnabledStoreWhenLoggingNotEnabled() throws Exception {
+        store = createStore(false, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put(new Windowed<>("a", new SessionWindow(0, 10)), "b");
+        assertTrue(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldReturnCachedSessionStoreWhenCachingEnabled() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        context.setTime(1);
+        store.put(new Windowed<>("a", new SessionWindow(0, 10)), "b");
+        store.put(new Windowed<>("b", new SessionWindow(0, 10)), "c");
+        assertThat(store, is(instanceOf(CachingSessionStore.class)));
+        assertThat(cache.size(), is(2L));
+    }
+
+    @Test
+    public void shouldReturnRocksDbStoreWhenCachingAndLoggingDisabled() throws Exception {
+        store = createStore(false, false);
+        assertThat(store, is(instanceOf(RocksDBSessionStore.class)));
+    }
+
+    @Test
+    public void shouldReturnRocksDbStoreWhenCachingDisabled() throws Exception {
+        store = createStore(true, false);
+        assertThat(store, is(instanceOf(RocksDBSessionStore.class)));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenCached() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenLogged() throws Exception {
+        store = createStore(true, false);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenNotLoggedOrCached() throws Exception {
+        store = createStore(false, false);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+
+
+    private SessionStore<String, String> createStore(final boolean logged, final boolean cached) {
+        return new RocksDBSessionStoreSupplier<>(STORE_NAME,
+                                                 10,
+                                                 Serdes.String(),
+                                                 Serdes.String(),
+                                                 logged,
+                                                 Collections.<String, String>emptyMap(),
+                                                 cached).get();
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java
new file mode 100644
index 0000000..897ec62
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java
@@ -0,0 +1,168 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.state.WindowStore;
+import org.apache.kafka.test.MockProcessorContext;
+import org.apache.kafka.test.NoOpRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class RocksDBWindowStoreSupplierTest {
+
+    private static final String STORE_NAME = "name";
+    private WindowStore<String, String> store;
+    private final ThreadCache cache = new ThreadCache("test", 1024, new MockStreamsMetrics(new Metrics()));
+    private final MockProcessorContext context = new MockProcessorContext(null,
+                                                                          TestUtils.tempDirectory(),
+                                                                          Serdes.String(),
+                                                                          Serdes.String(),
+                                                                          new NoOpRecordCollector(),
+                                                                          cache);
+
+    @After
+    public void close() {
+        store.close();
+    }
+
+    @Test
+    public void shouldCreateLoggingEnabledStoreWhenWindowStoreLogged() throws Exception {
+        store = createStore(true, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put("a", "b");
+        assertFalse(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldNotBeLoggingEnabledStoreWhenLogginNotEnabled() throws Exception {
+        store = createStore(false, false);
+        final List<ProducerRecord> logged = new ArrayList<>();
+        final NoOpRecordCollector collector = new NoOpRecordCollector() {
+            @Override
+            public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) {
+                logged.add(record);
+            }
+        };
+        final MockProcessorContext context = new MockProcessorContext(null,
+                                                                      TestUtils.tempDirectory(),
+                                                                      Serdes.String(),
+                                                                      Serdes.String(),
+                                                                      collector,
+                                                                      cache);
+        context.setTime(1);
+        store.init(context, store);
+        store.put("a", "b");
+        assertTrue(logged.isEmpty());
+    }
+
+    @Test
+    public void shouldBeCachedWindowStoreWhenCachingEnabled() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        context.setTime(1);
+        store.put("a", "b");
+        store.put("b", "c");
+        assertThat(store, is(instanceOf(CachingWindowStore.class)));
+        assertThat(context.getCache().size(), is(2L));
+    }
+
+    @Test
+    public void shouldReturnRocksDbStoreWhenCachingAndLoggingDisabled() throws Exception {
+        store = createStore(false, false);
+        assertThat(store, is(instanceOf(RocksDBWindowStore.class)));
+    }
+
+    @Test
+    public void shouldReturnRocksDbStoreWhenCachingDisabled() throws Exception {
+        store = createStore(true, false);
+        assertThat(store, is(instanceOf(RocksDBWindowStore.class)));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenCached() throws Exception {
+        store = createStore(false, true);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenLogged() throws Exception {
+        store = createStore(true, false);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldHaveMeteredStoreWhenNotLoggedOrCached() throws Exception {
+        store = createStore(false, false);
+        store.init(context, store);
+        final StreamsMetrics metrics = context.metrics();
+        assertFalse(metrics.metrics().isEmpty());
+    }
+
+    @SuppressWarnings("unchecked")
+    private WindowStore<String, String> createStore(final boolean logged, final boolean cached) {
+        return new RocksDBWindowStoreSupplier<>(STORE_NAME,
+                                                10,
+                                                3,
+                                                false,
+                                                Serdes.String(),
+                                                Serdes.String(),
+                                                10,
+                                                logged,
+                                                Collections.<String, String>emptyMap(),
+                                                cached).get();
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
index a522592..79223de 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
@@ -490,8 +490,6 @@ public class RocksDBWindowStoreTest {
                     recordCollector, cache);
 
             WindowStore<Integer, String> store = createWindowStore(context, false, true);
-            RocksDBWindowStore<Integer, String> inner =
-                    (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner();
             try {
                 // to validate segments
                 final Segments segments = new Segments(windowName, retentionPeriod, numSegments);

http://git-wip-us.apache.org/repos/asf/kafka/blob/73b7ae00/streams/src/test/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIteratorTest.java
new file mode 100644
index 0000000..8c0d2fe
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SerializedKeyValueIteratorTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.StateSerdes;
+import org.apache.kafka.test.KeyValueIteratorStub;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class SerializedKeyValueIteratorTest {
+
+    private final StateSerdes<String, String> serdes = new StateSerdes<>("blah", Serdes.String(), Serdes.String());
+    private final Iterator<KeyValue<Bytes, byte[]>> iterator
+            = Arrays.asList(KeyValue.pair(Bytes.wrap("hi".getBytes()), "there".getBytes()),
+                            KeyValue.pair(Bytes.wrap("hello".getBytes()), "world".getBytes()))
+            .iterator();
+    private final DelegatingPeekingKeyValueIterator<Bytes, byte[]> peeking
+            = new DelegatingPeekingKeyValueIterator<>("store", new KeyValueIteratorStub<>(iterator));
+    private final SerializedKeyValueIterator<String, String> serializedKeyValueIterator
+            = new SerializedKeyValueIterator<>(peeking, serdes);
+
+    @Test
+    public void shouldReturnTrueOnHasNextWhenMoreResults() {
+        assertTrue(serializedKeyValueIterator.hasNext());
+    }
+
+    @Test
+    public void shouldReturnNextValueWhenItExists() throws Exception {
+        assertThat(serializedKeyValueIterator.next(), equalTo(KeyValue.pair("hi", "there")));
+        assertThat(serializedKeyValueIterator.next(), equalTo(KeyValue.pair("hello", "world")));
+    }
+
+    @Test
+    public void shouldReturnFalseOnHasNextWhenNoMoreResults() throws Exception {
+        advanceIteratorToEnd();
+        assertFalse(serializedKeyValueIterator.hasNext());
+    }
+
+    @Test
+    public void shouldThrowNoSuchElementOnNextWhenIteratorExhausted() throws Exception {
+        advanceIteratorToEnd();
+        try {
+            serializedKeyValueIterator.next();
+            fail("Expected NoSuchElementException on exhausted iterator");
+        } catch (final NoSuchElementException nse) {
+            // pass
+        }
+    }
+
+    @Test
+    public void shouldPeekNextKey() throws Exception {
+        assertThat(serializedKeyValueIterator.peekNextKey(), equalTo("hi"));
+        serializedKeyValueIterator.next();
+        assertThat(serializedKeyValueIterator.peekNextKey(), equalTo("hello"));
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void shouldThrowUnsupportedOperationOnRemove() throws Exception {
+        serializedKeyValueIterator.remove();
+    }
+
+    private void advanceIteratorToEnd() {
+        serializedKeyValueIterator.next();
+        serializedKeyValueIterator.next();
+    }
+
+
+}
\ No newline at end of file


Mime
View raw message