kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From damian...@apache.org
Subject [1/2] kafka git commit: KAFKA-5650; add StateStoreBuilder interface and implementations
Date Thu, 07 Sep 2017 08:39:54 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 667cd60dc -> 9cbb9f093


http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java
new file mode 100644
index 0000000..a0500b6
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.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
+ *
+ *    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.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
+import org.apache.kafka.streams.state.WindowStore;
+
+import static org.apache.kafka.streams.state.internals.RocksDBWindowStoreSupplier.MIN_SEGMENTS;
+
+public class RocksDbWindowBytesStoreSupplier implements WindowBytesStoreSupplier {
+    private final String name;
+    private final long retentionPeriod;
+    private final int segments;
+    private final long windowSize;
+    private final boolean retainDuplicates;
+
+    public RocksDbWindowBytesStoreSupplier(final String name,
+                                           final long retentionPeriod,
+                                           final int segments,
+                                           final long windowSize,
+                                           final boolean retainDuplicates) {
+        if (segments < MIN_SEGMENTS) {
+            throw new IllegalArgumentException("numSegments must be >= " + MIN_SEGMENTS);
+        }
+        this.name = name;
+        this.retentionPeriod = retentionPeriod;
+        this.segments = segments;
+        this.windowSize = windowSize;
+        this.retainDuplicates = retainDuplicates;
+    }
+
+    @Override
+    public String name() {
+        return name;
+    }
+
+    @Override
+    public WindowStore<Bytes, byte[]> get() {
+        final RocksDBSegmentedBytesStore segmentedBytesStore = new RocksDBSegmentedBytesStore(
+                name,
+                retentionPeriod,
+                segments,
+                new WindowKeySchema()
+        );
+        return RocksDBWindowStore.bytesStore(segmentedBytesStore,
+                                             retainDuplicates,
+                                             windowSize);
+
+    }
+
+    @Override
+    public String metricsScope() {
+        return "rocksdb-window";
+    }
+
+    @Override
+    public int segments() {
+        return segments;
+    }
+
+    @Override
+    public long windowSize() {
+        return windowSize;
+    }
+
+    @Override
+    public boolean retainDuplicates() {
+        return retainDuplicates;
+    }
+
+    @Override
+    public long retentionPeriod() {
+        return retentionPeriod;
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilder.java
new file mode 100644
index 0000000..61919c3
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilder.java
@@ -0,0 +1,63 @@
+/*
+ * 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.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.SessionBytesStoreSupplier;
+import org.apache.kafka.streams.state.SessionStore;
+
+
+public class SessionStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, SessionStore<K,
V>> {
+
+    private final SessionBytesStoreSupplier storeSupplier;
+
+    public SessionStoreBuilder(final SessionBytesStoreSupplier storeSupplier,
+                               final Serde<K> keySerde,
+                               final Serde<V> valueSerde,
+                               final Time time) {
+        super(storeSupplier.name(), keySerde, valueSerde, time);
+        this.storeSupplier = storeSupplier;
+    }
+
+    @Override
+    public SessionStore<K, V> build() {
+        return new MeteredSessionStore<>(maybeWrapCaching(maybeWrapLogging(storeSupplier.get())),
+                                         storeSupplier.metricsScope(),
+                                         keySerde,
+                                         valueSerde,
+                                         time);
+    }
+
+    private SessionStore<Bytes, byte[]> maybeWrapCaching(final SessionStore<Bytes,
byte[]> inner) {
+        if (!enableCaching) {
+            return inner;
+        }
+        return new CachingSessionStore<>(inner,
+                                         keySerde,
+                                         valueSerde,
+                                         storeSupplier.segmentIntervalMs());
+    }
+
+    private SessionStore<Bytes, byte[]> maybeWrapLogging(final SessionStore<Bytes,
byte[]> inner) {
+        if (!enableLogging) {
+            return inner;
+        }
+        return new ChangeLoggingSessionBytesStore(inner);
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreBuilder.java
new file mode 100644
index 0000000..97b4883
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreBuilder.java
@@ -0,0 +1,68 @@
+/*
+ * 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.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.WindowBytesStoreSupplier;
+import org.apache.kafka.streams.state.WindowStore;
+
+
+public class WindowStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, WindowStore<K,
V>> {
+
+    private final WindowBytesStoreSupplier storeSupplier;
+
+    public WindowStoreBuilder(final WindowBytesStoreSupplier storeSupplier,
+                              final Serde<K> keySerde,
+                              final Serde<V> valueSerde,
+                              final Time time) {
+        super(storeSupplier.name(), keySerde, valueSerde, time);
+        this.storeSupplier = storeSupplier;
+    }
+
+    @Override
+    public WindowStore<K, V> build() {
+        return new MeteredWindowStore<>(maybeWrapCaching(maybeWrapLogging(storeSupplier.get())),
+                                        storeSupplier.metricsScope(),
+                                        time,
+                                        keySerde,
+                                        valueSerde);
+    }
+
+    private WindowStore<Bytes, byte[]> maybeWrapCaching(final WindowStore<Bytes,
byte[]> inner) {
+        if (!enableCaching) {
+            return inner;
+        }
+        return new CachingWindowStore<>(inner,
+                                        keySerde,
+                                        valueSerde,
+                                        storeSupplier.windowSize(),
+                                        storeSupplier.segments());
+    }
+
+    private WindowStore<Bytes, byte[]> maybeWrapLogging(final WindowStore<Bytes,
byte[]> inner) {
+        if (!enableLogging) {
+            return inner;
+        }
+        return new ChangeLoggingWindowBytesStore(inner, storeSupplier.retainDuplicates());
+    }
+
+    public long retentionPeriod() {
+        return storeSupplier.retentionPeriod();
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
index 12947d8..dbdc854 100644
--- a/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
@@ -23,10 +23,12 @@ 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.internals.InternalTopologyBuilder;
-import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
 import org.apache.kafka.test.MockProcessorSupplier;
 import org.apache.kafka.test.MockStateStoreSupplier;
 import org.apache.kafka.test.ProcessorTopologyTestDriver;
+import org.easymock.EasyMock;
 import org.junit.Test;
 
 import java.util.Arrays;
@@ -42,6 +44,8 @@ import static org.junit.Assert.fail;
 
 public class TopologyTest {
 
+    private final StoreBuilder storeBuilder = EasyMock.createNiceMock(StoreBuilder.class);
+    private final KeyValueStoreBuilder globalStoreBuilder = EasyMock.createNiceMock(KeyValueStoreBuilder.class);
     private final Topology topology = new Topology();
     private final InternalTopologyBuilder.TopologyDescription expectedDescription = new InternalTopologyBuilder.TopologyDescription();
 
@@ -203,38 +207,52 @@ public class TopologyTest {
 
     @Test(expected = TopologyException.class)
     public void shouldNotAllowToAddStateStoreToNonExistingProcessor() {
-        topology.addStateStore(new MockStateStoreSupplier("store", false), "no-such-processsor");
+        mockStoreBuilder();
+        EasyMock.replay(storeBuilder);
+        topology.addStateStore(storeBuilder, "no-such-processsor");
     }
 
     @Test
     public void shouldNotAllowToAddStateStoreToSource() {
+        mockStoreBuilder();
+        EasyMock.replay(storeBuilder);
         topology.addSource("source-1", "topic-1");
         try {
-            topology.addStateStore(new MockStateStoreSupplier("store", false), "source-1");
+            topology.addStateStore(storeBuilder, "source-1");
             fail("Should have thrown TopologyException for adding store to source node");
         } catch (final TopologyException expected) { }
     }
 
     @Test
     public void shouldNotAllowToAddStateStoreToSink() {
+        mockStoreBuilder();
+        EasyMock.replay(storeBuilder);
         topology.addSink("sink-1", "topic-1");
         try {
-            topology.addStateStore(new MockStateStoreSupplier("store", false), "sink-1");
+            topology.addStateStore(storeBuilder, "sink-1");
             fail("Should have thrown TopologyException for adding store to sink node");
         } catch (final TopologyException expected) { }
     }
 
+    private void mockStoreBuilder() {
+        EasyMock.expect(storeBuilder.name()).andReturn("store").anyTimes();
+        EasyMock.expect(storeBuilder.logConfig()).andReturn(Collections.emptyMap());
+        EasyMock.expect(storeBuilder.loggingEnabled()).andReturn(false);
+    }
+
     @Test
     public void shouldNotAllowToAddStoreWithSameName() {
-        topology.addStateStore(new MockStateStoreSupplier("store", false));
+        mockStoreBuilder();
+        EasyMock.replay(storeBuilder);
+        topology.addStateStore(storeBuilder);
         try {
-            topology.addStateStore(new MockStateStoreSupplier("store", false));
+            topology.addStateStore(storeBuilder);
             fail("Should have thrown TopologyException for duplicate store name");
         } catch (final TopologyException expected) { }
     }
 
     @Test(expected = TopologyBuilderException.class)
-    public void shouldThroughOnUnassignedStateStoreAccess() throws Exception {
+    public void shouldThrowOnUnassignedStateStoreAccess() throws Exception {
         final String sourceNodeName = "source";
         final String goodNodeName = "goodGuy";
         final String badNodeName = "badGuy";
@@ -243,12 +261,14 @@ public class TopologyTest {
         config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "host:1");
         config.put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
         final StreamsConfig streamsConfig = new StreamsConfig(config);
-
+        mockStoreBuilder();
+        EasyMock.expect(storeBuilder.build()).andReturn(new MockStateStoreSupplier.MockStateStore("store",
false));
+        EasyMock.replay(storeBuilder);
         topology
             .addSource(sourceNodeName, "topic")
             .addProcessor(goodNodeName, new LocalMockProcessorSupplier(), sourceNodeName)
             .addStateStore(
-                Stores.create(LocalMockProcessorSupplier.STORE_NAME).withStringKeys().withStringValues().inMemory().build(),
+                storeBuilder,
                 goodNodeName)
             .addProcessor(badNodeName, new LocalMockProcessorSupplier(), sourceNodeName);
 
@@ -292,8 +312,10 @@ public class TopologyTest {
 
     @Test(expected = TopologyException.class)
     public void shouldNotAllowToAddGlobalStoreWithSourceNameEqualsProcessorName() {
+        EasyMock.expect(globalStoreBuilder.name()).andReturn("anyName").anyTimes();
+        EasyMock.replay(globalStoreBuilder);
         topology.addGlobalStore(
-            new MockStateStoreSupplier("anyName", false, false),
+            globalStoreBuilder,
             "sameName",
             null,
             null,
@@ -611,7 +633,10 @@ public class TopologyTest {
         topology.addProcessor(processorName, new MockProcessorSupplier(), parentNames);
         if (newStores) {
             for (final String store : storeNames) {
-                topology.addStateStore(new MockStateStoreSupplier(store, false), processorName);
+                final StoreBuilder storeBuilder = EasyMock.createNiceMock(StoreBuilder.class);
+                EasyMock.expect(storeBuilder.name()).andReturn(store).anyTimes();
+                EasyMock.replay(storeBuilder);
+                topology.addStateStore(storeBuilder, processorName);
             }
         } else {
             topology.connectProcessorAndStateStores(processorName, storeNames);
@@ -651,8 +676,11 @@ public class TopologyTest {
                                                                 final String sourceName,
                                                                 final String globalTopicName,
                                                                 final String processorName)
{
+        final KeyValueStoreBuilder globalStoreBuilder = EasyMock.createNiceMock(KeyValueStoreBuilder.class);
+        EasyMock.expect(globalStoreBuilder.name()).andReturn(globalStoreName).anyTimes();
+        EasyMock.replay(globalStoreBuilder);
         topology.addGlobalStore(
-            new MockStateStoreSupplier(globalStoreName, false, false),
+            globalStoreBuilder,
             sourceName,
             null,
             null,

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
index 9bd8756..91edac5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
@@ -470,7 +470,7 @@ public class InternalTopologyBuilderTest {
 
     @Test(expected = NullPointerException.class)
     public void shouldNotAddNullStateStoreSupplier() throws Exception {
-        builder.addStateStore(null);
+        builder.addStateStore((StateStoreSupplier) null);
     }
 
     private Set<String> nodeNames(final Collection<ProcessorNode> nodes) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
index 66adbf5..700b243 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
@@ -18,11 +18,20 @@ package org.apache.kafka.streams.state;
 
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.processor.StateStoreSupplier;
+import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore;
+import org.apache.kafka.streams.state.internals.MemoryNavigableLRUCache;
+import org.apache.kafka.streams.state.internals.RocksDBSessionStore;
+import org.apache.kafka.streams.state.internals.RocksDBStore;
+import org.apache.kafka.streams.state.internals.RocksDBWindowStore;
 import org.junit.Test;
 
 import java.util.Collections;
 import java.util.Map;
 
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.hamcrest.core.IsNot.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -30,6 +39,7 @@ import static org.junit.Assert.fail;
 
 public class StoresTest {
 
+    @SuppressWarnings("deprecation")
     @Test
     public void shouldCreateInMemoryStoreSupplierWithLoggedConfig() throws Exception {
         final StateStoreSupplier supplier = Stores.create("store")
@@ -44,6 +54,7 @@ public class StoresTest {
         assertEquals("1000", config.get("retention.ms"));
     }
 
+    @SuppressWarnings("deprecation")
     @Test
     public void shouldCreateInMemoryStoreSupplierNotLogged() throws Exception {
         final StateStoreSupplier supplier = Stores.create("store")
@@ -56,6 +67,7 @@ public class StoresTest {
         assertFalse(supplier.loggingEnabled());
     }
 
+    @SuppressWarnings("deprecation")
     @Test
     public void shouldCreatePersistenStoreSupplierWithLoggedConfig() throws Exception {
         final StateStoreSupplier supplier = Stores.create("store")
@@ -70,6 +82,7 @@ public class StoresTest {
         assertEquals("1000", config.get("retention.ms"));
     }
 
+    @SuppressWarnings("deprecation")
     @Test
     public void shouldCreatePersistenStoreSupplierNotLogged() throws Exception {
         final StateStoreSupplier supplier = Stores.create("store")
@@ -95,4 +108,53 @@ public class StoresTest {
          // ok
         }
     }
+
+    @Test
+    public void shouldCreateInMemoryKeyValueStore() {
+        assertThat(Stores.inMemoryKeyValueStore("memory").get(), instanceOf(InMemoryKeyValueStore.class));
+    }
+
+    @Test
+    public void shouldCreateMemoryNavigableCache() {
+        assertThat(Stores.lruMap("map", 10).get(), instanceOf(MemoryNavigableLRUCache.class));
+    }
+
+    @Test
+    public void shouldCreateRocksDbStore() {
+        assertThat(Stores.persistentKeyValueStore("store").get(), instanceOf(RocksDBStore.class));
+    }
+
+    @Test
+    public void shouldCreateRocksDbWindowStore() {
+        assertThat(Stores.persistentWindowStore("store", 1, 3, 1, false).get(), instanceOf(RocksDBWindowStore.class));
+    }
+
+    @Test
+    public void shouldCreateRocksDbSessionStore() {
+        assertThat(Stores.persistentSessionStore("store", 1).get(), instanceOf(RocksDBSessionStore.class));
+    }
+
+    @Test
+    public void shouldBuildWindowStore() {
+        final WindowStore<String, String> store = Stores.windowStoreBuilder(Stores.persistentWindowStore("store",
3, 2, 3, true),
+                                                                      Serdes.String(),
+                                                                      Serdes.String()).build();
+        assertThat(store, not(nullValue()));
+    }
+
+    @Test
+    public void shouldBuildKeyValueStore() {
+        final KeyValueStore<String, String> store = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("name"),
+                                                                          Serdes.String(),
+                                                                          Serdes.String()).build();
+        assertThat(store, not(nullValue()));
+    }
+
+    @Test
+    public void shouldBuildSessionStore() {
+        final SessionStore<String, String> store = Stores.sessionStoreBuilder(Stores.persistentSessionStore("name",
10),
+                                                                       Serdes.String(),
+                                                                       Serdes.String()).build();
+        assertThat(store, not(nullValue()));
+    }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilderTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilderTest.java
new file mode 100644
index 0000000..2d378d8
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilderTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.hamcrest.CoreMatchers;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+
+@RunWith(EasyMockRunner.class)
+public class KeyValueStoreBuilderTest {
+
+    @Mock(type = MockType.NICE)
+    private KeyValueBytesStoreSupplier supplier;
+    @Mock(type = MockType.NICE)
+    private KeyValueStore<Bytes, byte[]> inner;
+    private KeyValueStoreBuilder<String, String> builder;
+
+    @Before
+    public void setUp() throws Exception {
+        EasyMock.expect(supplier.get()).andReturn(inner);
+        EasyMock.expect(supplier.name()).andReturn("name");
+        EasyMock.replay(supplier);
+        builder = new KeyValueStoreBuilder<>(supplier,
+                                             Serdes.String(),
+                                             Serdes.String(),
+                                             new MockTime()
+        );
+
+    }
+
+    @Test
+    public void shouldHaveMeteredStoreAsOuterStore() {
+        final KeyValueStore<String, String> store = builder.build();
+        assertThat(store, instanceOf(MeteredKeyValueBytesStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreByDefault() {
+        final KeyValueStore<String, String> store = builder.build();
+        assertThat(store, instanceOf(MeteredKeyValueBytesStore.class));
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, instanceOf(ChangeLoggingKeyValueBytesStore.class));
+    }
+
+    @Test
+    public void shouldNotHaveChangeLoggingStoreWhenDisabled() {
+        final KeyValueStore<String, String> store = builder.withLoggingDisabled().build();
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingStoreWhenEnabled() {
+        final KeyValueStore<String, String> store = builder.withCachingEnabled().build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredKeyValueBytesStore.class));
+        assertThat(wrapped, instanceOf(CachingKeyValueStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() {
+        final KeyValueStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredKeyValueBytesStore.class));
+        assertThat(wrapped, instanceOf(ChangeLoggingKeyValueBytesStore.class));
+        assertThat(((WrappedStateStore) wrapped).wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingAndChangeLoggingWhenBothEnabled() {
+        final KeyValueStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .withCachingEnabled()
+                .build();
+        final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrappedStore();
+        final WrappedStateStore changeLogging = (WrappedStateStore) caching.wrappedStore();
+        assertThat(store, instanceOf(MeteredKeyValueBytesStore.class));
+        assertThat(caching, instanceOf(CachingKeyValueStore.class));
+        assertThat(changeLogging, instanceOf(ChangeLoggingKeyValueBytesStore.class));
+        assertThat(changeLogging.wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfInnerIsNull() {
+        new KeyValueStoreBuilder<>(null, Serdes.String(), Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfKeySerdeIsNull() {
+        new KeyValueStoreBuilder<>(supplier, null, Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfValueSerdeIsNull() {
+        new KeyValueStoreBuilder<>(supplier, Serdes.String(), null, new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfTimeIsNull() {
+        new KeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfMetricsScopeIsNull() {
+        new KeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), new
MockTime());
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderTest.java
new file mode 100644
index 0000000..621a1c2
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SessionStoreBuilderTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
+import org.apache.kafka.streams.state.SessionStore;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.hamcrest.CoreMatchers;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+
+@RunWith(EasyMockRunner.class)
+public class SessionStoreBuilderTest {
+
+    @Mock(type = MockType.NICE)
+    private SessionBytesStoreSupplier supplier;
+    @Mock(type = MockType.NICE)
+    private SessionStore<Bytes, byte[]> inner;
+    private SessionStoreBuilder<String, String> builder;
+
+    @Before
+    public void setUp() throws Exception {
+
+        EasyMock.expect(supplier.get()).andReturn(inner);
+        EasyMock.expect(supplier.name()).andReturn("name");
+        EasyMock.replay(supplier);
+
+        builder = new SessionStoreBuilder<>(supplier,
+                                            Serdes.String(),
+                                            Serdes.String(),
+                                            new MockTime()
+        );
+    }
+
+    @Test
+    public void shouldHaveMeteredStoreAsOuterStore() {
+        final SessionStore<String, String> store = builder.build();
+        assertThat(store, instanceOf(MeteredSessionStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreByDefault() {
+        final SessionStore<String, String> store = builder.build();
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, instanceOf(ChangeLoggingSessionBytesStore.class));
+    }
+
+    @Test
+    public void shouldNotHaveChangeLoggingStoreWhenDisabled() {
+        final SessionStore<String, String> store = builder.withLoggingDisabled().build();
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingStoreWhenEnabled() {
+        final SessionStore<String, String> store = builder.withCachingEnabled().build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredSessionStore.class));
+        assertThat(wrapped, instanceOf(CachingSessionStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() {
+        final SessionStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredSessionStore.class));
+        assertThat(wrapped, instanceOf(ChangeLoggingSessionBytesStore.class));
+        assertThat(((WrappedStateStore) wrapped).wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingAndChangeLoggingWhenBothEnabled() {
+        final SessionStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .withCachingEnabled()
+                .build();
+        final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrappedStore();
+        final WrappedStateStore changeLogging = (WrappedStateStore) caching.wrappedStore();
+        assertThat(store, instanceOf(MeteredSessionStore.class));
+        assertThat(caching, instanceOf(CachingSessionStore.class));
+        assertThat(changeLogging, instanceOf(ChangeLoggingSessionBytesStore.class));
+        assertThat(changeLogging.wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfInnerIsNull() {
+        new SessionStoreBuilder<>(null, Serdes.String(), Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfKeySerdeIsNull() {
+        new SessionStoreBuilder<>(supplier, null, Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfValueSerdeIsNull() {
+        new SessionStoreBuilder<>(supplier, Serdes.String(), null, new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfTimeIsNull() {
+        new SessionStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfMetricsScopeIsNull() {
+        new SessionStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), new MockTime());
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/9cbb9f09/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreBuilderTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreBuilderTest.java
new file mode 100644
index 0000000..25b8178
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreBuilderTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
+import org.apache.kafka.streams.state.WindowStore;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.hamcrest.CoreMatchers;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+
+@RunWith(EasyMockRunner.class)
+public class WindowStoreBuilderTest {
+
+    @Mock(type = MockType.NICE)
+    private WindowBytesStoreSupplier supplier;
+    @Mock(type = MockType.NICE)
+    private WindowStore<Bytes, byte[]> inner;
+    private WindowStoreBuilder<String, String> builder;
+
+    @Before
+    public void setUp() throws Exception {
+        EasyMock.expect(supplier.get()).andReturn(inner);
+        EasyMock.expect(supplier.name()).andReturn("name");
+        EasyMock.replay(supplier);
+
+        builder = new WindowStoreBuilder<>(supplier,
+                                           Serdes.String(),
+                                           Serdes.String(),
+                                           new MockTime());
+
+    }
+
+    @Test
+    public void shouldHaveMeteredStoreAsOuterStore() {
+        final WindowStore<String, String> store = builder.build();
+        assertThat(store, instanceOf(MeteredWindowStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreByDefault() {
+        final WindowStore<String, String> store = builder.build();
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, instanceOf(ChangeLoggingWindowBytesStore.class));
+    }
+
+    @Test
+    public void shouldNotHaveChangeLoggingStoreWhenDisabled() {
+        final WindowStore<String, String> store = builder.withLoggingDisabled().build();
+        final StateStore next = ((WrappedStateStore) store).wrappedStore();
+        assertThat(next, CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingStoreWhenEnabled() {
+        final WindowStore<String, String> store = builder.withCachingEnabled().build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredWindowStore.class));
+        assertThat(wrapped, instanceOf(CachingWindowStore.class));
+    }
+
+    @Test
+    public void shouldHaveChangeLoggingStoreWhenLoggingEnabled() {
+        final WindowStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .build();
+        final StateStore wrapped = ((WrappedStateStore) store).wrappedStore();
+        assertThat(store, instanceOf(MeteredWindowStore.class));
+        assertThat(wrapped, instanceOf(ChangeLoggingWindowBytesStore.class));
+        assertThat(((WrappedStateStore) wrapped).wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test
+    public void shouldHaveCachingAndChangeLoggingWhenBothEnabled() {
+        final WindowStore<String, String> store = builder
+                .withLoggingEnabled(Collections.<String, String>emptyMap())
+                .withCachingEnabled()
+                .build();
+        final WrappedStateStore caching = (WrappedStateStore) ((WrappedStateStore) store).wrappedStore();
+        final WrappedStateStore changeLogging = (WrappedStateStore) caching.wrappedStore();
+        assertThat(store, instanceOf(MeteredWindowStore.class));
+        assertThat(caching, instanceOf(CachingWindowStore.class));
+        assertThat(changeLogging, instanceOf(ChangeLoggingWindowBytesStore.class));
+        assertThat(changeLogging.wrappedStore(), CoreMatchers.<StateStore>equalTo(inner));
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfInnerIsNull() {
+        new WindowStoreBuilder<>(null, Serdes.String(), Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfKeySerdeIsNull() {
+        new WindowStoreBuilder<>(supplier, null, Serdes.String(), new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfValueSerdeIsNull() {
+        new WindowStoreBuilder<>(supplier, Serdes.String(), null, new MockTime());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void shouldThrowNullPointerIfTimeIsNull() {
+        new WindowStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), null);
+    }
+
+}
\ No newline at end of file


Mime
View raw message