You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/02/03 19:12:54 UTC
[1/5] kafka git commit: KAFKA-3452 Follow-up: Optimize ByteStore
Scenarios
Repository: kafka
Updated Branches:
refs/heads/trunk d95f22c12 -> 7ebc5da60
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
index 499d823..7b112e0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
@@ -52,17 +52,17 @@ public class ThreadCacheTest {
toInsert.size() * memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""),
new MockStreamsMetrics(new Metrics()));
- for (int i = 0; i < toInsert.size(); i++) {
- byte[] key = toInsert.get(i).key.getBytes();
- byte[] value = toInsert.get(i).value.getBytes();
+ for (KeyValue<String, String> kvToInsert : toInsert) {
+ Bytes key = Bytes.wrap(kvToInsert.key.getBytes());
+ byte[] value = kvToInsert.value.getBytes();
cache.put(name, key, new LRUCacheEntry(value, true, 1L, 1L, 1, ""));
}
- for (int i = 0; i < toInsert.size(); i++) {
- byte[] key = toInsert.get(i).key.getBytes();
+ for (KeyValue<String, String> kvToInsert : toInsert) {
+ Bytes key = Bytes.wrap(kvToInsert.key.getBytes());
LRUCacheEntry entry = cache.get(name, key);
- assertEquals(entry.isDirty, true);
- assertEquals(new String(entry.value), toInsert.get(i).value);
+ assertEquals(entry.isDirty(), true);
+ assertEquals(new String(entry.value), kvToInsert.value);
}
assertEquals(cache.gets(), 5);
assertEquals(cache.puts(), 5);
@@ -73,10 +73,8 @@ public class ThreadCacheTest {
private void checkOverheads(double entryFactor, double systemFactor, long desiredCacheSize, int keySizeBytes,
int valueSizeBytes) {
Runtime runtime = Runtime.getRuntime();
- byte[] key = new byte[keySizeBytes];
- byte[] value = new byte[valueSizeBytes];
final String name = "name";
- long numElements = desiredCacheSize / memoryCacheEntrySize(key, value, "");
+ long numElements = desiredCacheSize / memoryCacheEntrySize(new byte[keySizeBytes], new byte[valueSizeBytes], "");
System.gc();
long prevRuntimeMemory = runtime.totalMemory() - runtime.freeMemory();
@@ -86,8 +84,8 @@ public class ThreadCacheTest {
assertEquals(size, 0);
for (int i = 0; i < numElements; i++) {
String keyStr = "K" + i;
- key = keyStr.getBytes();
- value = new byte[valueSizeBytes];
+ Bytes key = Bytes.wrap(keyStr.getBytes());
+ byte[] value = new byte[valueSizeBytes];
cache.put(name, key, new LRUCacheEntry(value, true, 1L, 1L, 1, ""));
}
@@ -144,7 +142,7 @@ public class ThreadCacheTest {
@Test
public void evict() throws IOException {
final List<KeyValue<String, String>> received = new ArrayList<>();
- List<KeyValue<String, String>> expected = Arrays.asList(
+ List<KeyValue<String, String>> expected = Collections.singletonList(
new KeyValue<>("K1", "V1"));
List<KeyValue<String, String>> toInsert = Arrays.asList(
@@ -168,10 +166,9 @@ public class ThreadCacheTest {
});
-
- for (int i = 0; i < toInsert.size(); i++) {
- byte[] key = toInsert.get(i).key.getBytes();
- byte[] value = toInsert.get(i).value.getBytes();
+ for (KeyValue<String, String> kvToInsert : toInsert) {
+ final Bytes key = Bytes.wrap(kvToInsert.key.getBytes());
+ final byte[] value = kvToInsert.value.getBytes();
cache.put(namespace, key, new LRUCacheEntry(value, true, 1, 1, 1, ""));
}
@@ -186,16 +183,16 @@ public class ThreadCacheTest {
@Test
public void shouldDelete() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final byte[] key = new byte[]{0};
+ final Bytes key = Bytes.wrap(new byte[]{0});
- cache.put("name", key, dirtyEntry(key));
- assertEquals(key, cache.delete("name", key).value);
+ cache.put("name", key, dirtyEntry(key.get()));
+ assertEquals(key.get(), cache.delete("name", key).value);
assertNull(cache.get("name", key));
}
@Test
public void shouldNotFlushAfterDelete() throws Exception {
- final byte[] key = new byte[]{0};
+ final Bytes key = Bytes.wrap(new byte[]{0});
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final String namespace = "namespace";
@@ -205,8 +202,8 @@ public class ThreadCacheTest {
received.addAll(dirty);
}
});
- cache.put(namespace, key, dirtyEntry(key));
- assertEquals(key, cache.delete(namespace, key).value);
+ cache.put(namespace, key, dirtyEntry(key.get()));
+ assertEquals(key.get(), cache.delete(namespace, key).value);
// flushing should have no further effect
cache.flush(namespace);
@@ -216,63 +213,63 @@ public class ThreadCacheTest {
@Test
public void shouldNotBlowUpOnNonExistentKeyWhenDeleting() throws Exception {
+ final Bytes key = Bytes.wrap(new byte[]{0});
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final byte[] key = new byte[]{0};
- cache.put("name", key, dirtyEntry(key));
- assertNull(cache.delete("name", new byte[]{1}));
+ cache.put("name", key, dirtyEntry(key.get()));
+ assertNull(cache.delete("name", Bytes.wrap(new byte[]{1})));
}
@Test
public void shouldNotBlowUpOnNonExistentNamespaceWhenDeleting() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- assertNull(cache.delete("name", new byte[]{1}));
+ assertNull(cache.delete("name", Bytes.wrap(new byte[]{1})));
}
@Test
public void shouldNotClashWithOverlappingNames() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final byte[] nameByte = new byte[]{0};
- final byte[] name1Byte = new byte[]{1};
- cache.put("name", nameByte, dirtyEntry(nameByte));
- cache.put("name1", nameByte, dirtyEntry(name1Byte));
+ final Bytes nameByte = Bytes.wrap(new byte[]{0});
+ final Bytes name1Byte = Bytes.wrap(new byte[]{1});
+ cache.put("name", nameByte, dirtyEntry(nameByte.get()));
+ cache.put("name1", nameByte, dirtyEntry(name1Byte.get()));
- assertArrayEquals(nameByte, cache.get("name", nameByte).value);
- assertArrayEquals(name1Byte, cache.get("name1", nameByte).value);
+ assertArrayEquals(nameByte.get(), cache.get("name", nameByte).value);
+ assertArrayEquals(name1Byte.get(), cache.get("name1", nameByte).value);
}
@Test
public void shouldPeekNextKey() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final byte[] theByte = {0};
+ final Bytes theByte = Bytes.wrap(new byte[]{0});
final String namespace = "streams";
- cache.put(namespace, theByte, dirtyEntry(theByte));
- final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, theByte, new byte[]{1});
- assertEquals(Bytes.wrap(theByte), iterator.peekNextKey());
- assertEquals(Bytes.wrap(theByte), iterator.peekNextKey());
+ cache.put(namespace, theByte, dirtyEntry(theByte.get()));
+ final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, theByte, Bytes.wrap(new byte[]{1}));
+ assertEquals(theByte, iterator.peekNextKey());
+ assertEquals(theByte, iterator.peekNextKey());
}
@Test
public void shouldGetSameKeyAsPeekNext() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final byte[] theByte = {0};
+ final Bytes theByte = Bytes.wrap(new byte[]{0});
final String namespace = "streams";
- cache.put(namespace, theByte, dirtyEntry(theByte));
- final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, theByte, new byte[]{1});
+ cache.put(namespace, theByte, dirtyEntry(theByte.get()));
+ final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, theByte, Bytes.wrap(new byte[]{1}));
assertEquals(iterator.peekNextKey(), iterator.next().key);
}
@Test(expected = NoSuchElementException.class)
public void shouldThrowIfNoPeekNextKey() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", new byte[]{0}, new byte[]{1});
+ final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", Bytes.wrap(new byte[]{0}), Bytes.wrap(new byte[]{1}));
iterator.peekNextKey();
}
@Test
public void shouldReturnFalseIfNoNextKey() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
- final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", new byte[]{0}, new byte[]{1});
+ final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range("", Bytes.wrap(new byte[]{0}), Bytes.wrap(new byte[]{1}));
assertFalse(iterator.hasNext());
}
@@ -282,9 +279,9 @@ public class ThreadCacheTest {
final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}};
final String namespace = "streams";
for (final byte[] aByte : bytes) {
- cache.put(namespace, aByte, dirtyEntry(aByte));
+ cache.put(namespace, Bytes.wrap(aByte), dirtyEntry(aByte));
}
- final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, new byte[]{1}, new byte[]{4});
+ final ThreadCache.MemoryLRUCacheBytesIterator iterator = cache.range(namespace, Bytes.wrap(new byte[]{1}), Bytes.wrap(new byte[]{4}));
int bytesIndex = 1;
while (iterator.hasNext()) {
Bytes peekedKey = iterator.peekNextKey();
@@ -309,13 +306,13 @@ public class ThreadCacheTest {
});
byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}};
for (int i = 0; i < 5; i++) {
- cache.put(namespace, bytes[i], dirtyEntry(bytes[i]));
+ cache.put(namespace, Bytes.wrap(bytes[i]), dirtyEntry(bytes[i]));
}
assertEquals(5, cache.size());
- final ThreadCache.MemoryLRUCacheBytesIterator range = cache.range(namespace, new byte[]{0}, new byte[]{5});
+ final ThreadCache.MemoryLRUCacheBytesIterator range = cache.range(namespace, Bytes.wrap(new byte[]{0}), Bytes.wrap(new byte[]{5}));
// should evict byte[] {0}
- cache.put(namespace, new byte[]{6}, dirtyEntry(new byte[]{6}));
+ cache.put(namespace, Bytes.wrap(new byte[]{6}), dirtyEntry(new byte[]{6}));
assertEquals(Bytes.wrap(new byte[]{1}), range.peekNextKey());
}
@@ -334,9 +331,9 @@ public class ThreadCacheTest {
});
final List<byte[]> expected = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2});
for (byte[] bytes : expected) {
- cache.put("1", bytes, dirtyEntry(bytes));
+ cache.put("1", Bytes.wrap(bytes), dirtyEntry(bytes));
}
- cache.put("2", new byte[]{4}, dirtyEntry(new byte[]{4}));
+ cache.put("2", Bytes.wrap(new byte[]{4}), dirtyEntry(new byte[]{4}));
cache.flush("1");
assertEquals(expected, received);
@@ -356,9 +353,9 @@ public class ThreadCacheTest {
});
final List<byte[]> toInsert = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2});
for (byte[] bytes : toInsert) {
- cache.put("1", bytes, cleanEntry(bytes));
+ cache.put("1", Bytes.wrap(bytes), cleanEntry(bytes));
}
- cache.put("2", new byte[]{4}, cleanEntry(new byte[]{4}));
+ cache.put("2", Bytes.wrap(new byte[]{4}), cleanEntry(new byte[]{4}));
cache.flush("1");
assertEquals(Collections.EMPTY_LIST, received);
@@ -374,7 +371,7 @@ public class ThreadCacheTest {
received.addAll(dirty);
}
});
- cache.put(namespace, new byte[]{0}, dirtyEntry(new byte[]{0}));
+ cache.put(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{0}));
assertEquals(1, received.size());
// flushing should have no further effect
@@ -406,21 +403,22 @@ public class ThreadCacheTest {
}
});
- cache.putAll(namespace, Arrays.asList(KeyValue.pair(new byte[]{0}, dirtyEntry(new byte[]{5})),
- KeyValue.pair(new byte[]{1}, dirtyEntry(new byte[]{6}))));
+ cache.putAll(namespace, Arrays.asList(KeyValue.pair(Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5})),
+ KeyValue.pair(Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}))));
assertEquals(cache.evicts(), 2);
+ assertEquals(received.size(), 2);
}
@Test
public void shouldPutAll() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
- cache.putAll("name", Arrays.asList(KeyValue.pair(new byte[]{0}, dirtyEntry(new byte[]{5})),
- KeyValue.pair(new byte[]{1}, dirtyEntry(new byte[]{6}))));
+ cache.putAll("name", Arrays.asList(KeyValue.pair(Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5})),
+ KeyValue.pair(Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}))));
- assertArrayEquals(new byte[]{5}, cache.get("name", new byte[]{0}).value);
- assertArrayEquals(new byte[]{6}, cache.get("name", new byte[]{1}).value);
+ assertArrayEquals(new byte[]{5}, cache.get("name", Bytes.wrap(new byte[]{0})).value);
+ assertArrayEquals(new byte[]{6}, cache.get("name", Bytes.wrap(new byte[]{1})).value);
}
@Test
@@ -433,13 +431,13 @@ public class ThreadCacheTest {
received.addAll(dirty);
}
});
- cache.put("name", new byte[] {1}, cleanEntry(new byte[]{0}));
+ cache.put("name", Bytes.wrap(new byte[]{1}), cleanEntry(new byte[]{0}));
assertEquals(0, received.size());
}
@Test
public void shouldPutIfAbsent() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
- final byte[] key = {10};
+ final Bytes key = Bytes.wrap(new byte[]{10});
final byte[] value = {30};
assertNull(cache.putIfAbsent("n", key, dirtyEntry(value)));
assertArrayEquals(value, cache.putIfAbsent("n", key, dirtyEntry(new byte[]{8})).value);
@@ -458,11 +456,12 @@ public class ThreadCacheTest {
}
});
- cache.putIfAbsent(namespace, new byte[]{0}, dirtyEntry(new byte[]{5}));
- cache.putIfAbsent(namespace, new byte[]{1}, dirtyEntry(new byte[]{6}));
- cache.putIfAbsent(namespace, new byte[]{1}, dirtyEntry(new byte[]{6}));
+ cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5}));
+ cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}));
+ cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}));
assertEquals(cache.evicts(), 3);
+ assertEquals(received.size(), 3);
}
@Test
@@ -475,7 +474,7 @@ public class ThreadCacheTest {
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
// put an item into an empty cache when the total cache size
// is already > than maxCacheSizeBytes
- threadCache.put("other", new byte[]{0}, dirtyEntry(new byte[2]));
+ threadCache.put("other", Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[2]));
}
});
threadCache.addDirtyEntryFlushListener("other", new ThreadCache.DirtyEntryFlushListener() {
@@ -491,29 +490,29 @@ public class ThreadCacheTest {
}
});
- threadCache.put("another", new byte[]{1}, dirtyEntry(new byte[1]));
- threadCache.put("name", new byte[]{1}, dirtyEntry(new byte[1]));
+ threadCache.put("another", Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1]));
+ threadCache.put("name", Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1]));
// Put a large item such that when the eldest item is removed
// cache sizeInBytes() > maxCacheSizeBytes
int remaining = (int) (maxCacheSizeInBytes - threadCache.sizeBytes());
- threadCache.put("name", new byte[]{2}, dirtyEntry(new byte[remaining + 100]));
+ threadCache.put("name", Bytes.wrap(new byte[]{2}), dirtyEntry(new byte[remaining + 100]));
}
@Test
public void shouldCleanupNamedCacheOnClose() throws Exception {
final ThreadCache cache = new ThreadCache("testCache", 100000, new MockStreamsMetrics(new Metrics()));
- cache.put("one", new byte[]{1}, cleanEntry(new byte[] {1}));
- cache.put("two", new byte[]{1}, cleanEntry(new byte[] {1}));
+ cache.put("one", Bytes.wrap(new byte[]{1}), cleanEntry(new byte[] {1}));
+ cache.put("two", Bytes.wrap(new byte[]{1}), cleanEntry(new byte[] {1}));
assertEquals(cache.size(), 2);
cache.close("two");
assertEquals(cache.size(), 1);
- assertNull(cache.get("two", new byte[] {1}));
+ assertNull(cache.get("two", Bytes.wrap(new byte[]{1})));
}
@Test
public void shouldReturnNullIfKeyIsNull() throws Exception {
final ThreadCache threadCache = new ThreadCache("testCache", 10, new MockStreamsMetrics(new Metrics()));
- threadCache.put("one", new byte[]{1}, cleanEntry(new byte[] {1}));
+ threadCache.put("one", Bytes.wrap(new byte[]{1}), cleanEntry(new byte[] {1}));
assertNull(threadCache.get("one", null));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreUtilsTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreUtilsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreUtilsTest.java
index e0cb3ae..3a40b44 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreUtilsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/WindowStoreUtilsTest.java
@@ -19,6 +19,7 @@
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.state.StateSerdes;
import org.junit.Test;
@@ -32,13 +33,12 @@ public class WindowStoreUtilsTest {
final String key = "key1";
final long timestamp = 99L;
final int seqNum = 3;
- byte[] bytes = WindowStoreUtils.toBinaryKey(key, timestamp, seqNum, serdes);
- final String parsedKey = WindowStoreUtils.keyFromBinaryKey(bytes, serdes);
- final long parsedTs = WindowStoreUtils.timestampFromBinaryKey(bytes);
- final int parsedSeqNum = WindowStoreUtils.sequenceNumberFromBinaryKey(bytes);
+ Bytes bytes = WindowStoreUtils.toBinaryKey(key, timestamp, seqNum, serdes);
+ final String parsedKey = WindowStoreUtils.keyFromBinaryKey(bytes.get(), serdes);
+ final long parsedTs = WindowStoreUtils.timestampFromBinaryKey(bytes.get());
+ final int parsedSeqNum = WindowStoreUtils.sequenceNumberFromBinaryKey(bytes.get());
assertEquals(key, parsedKey);
assertEquals(timestamp, parsedTs);
assertEquals(seqNum, parsedSeqNum);
}
-
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/test/InMemoryKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/InMemoryKeyValueStore.java b/streams/src/test/java/org/apache/kafka/test/InMemoryKeyValueStore.java
deleted file mode 100644
index fc9b32b..0000000
--- a/streams/src/test/java/org/apache/kafka/test/InMemoryKeyValueStore.java
+++ /dev/null
@@ -1,152 +0,0 @@
-/**
- * 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.test;
-
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.state.KeyValueIterator;
-import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.internals.DelegatingPeekingKeyValueIterator;
-
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.TreeMap;
-
-public class InMemoryKeyValueStore<K, V> implements KeyValueStore<K, V> {
- private final TreeMap<K, V> map = new TreeMap<>();
- private final String name;
- private boolean open = true;
-
- public InMemoryKeyValueStore(final String name) {
- this.name = name;
- }
-
- @Override
- public void put(final K key, final V value) {
- map.put(key, value);
- }
-
- @Override
- public V putIfAbsent(final K key, final V value) {
- V orig = map.get(key);
- if (orig == null) {
- map.put(key, value);
- }
- return orig;
- }
-
- @Override
- public void putAll(final List<KeyValue<K, V>> entries) {
- for (KeyValue<K, V> entry : entries) {
- map.put(entry.key, entry.value);
- }
- }
-
- @Override
- public V delete(final K key) {
- return map.remove(key);
- }
-
- @Override
- public long approximateNumEntries() {
- return map.size();
- }
-
- @Override
- public String name() {
- return name;
- }
-
- @Override
- public void init(final ProcessorContext context, final StateStore root) {
- // no-op
- }
-
- @Override
- public void flush() {
- //no-op
- }
-
- @Override
- public void close() {
- open = false;
- }
-
- @Override
- public boolean persistent() {
- return false;
- }
-
- @Override
- public boolean isOpen() {
- return open;
- }
-
- @Override
- public V get(final K key) {
- return map.get(key);
- }
-
- @Override
- public KeyValueIterator<K, V> range(final K from, final K to) {
- return new DelegatingPeekingKeyValueIterator<>(name, new TheIterator(this.map.subMap(from, true, to, true).entrySet().iterator()));
- }
-
- @Override
- public KeyValueIterator<K, V> all() {
- return new DelegatingPeekingKeyValueIterator<>(name(), new TheIterator(map.entrySet().iterator()));
- }
-
- private class TheIterator implements KeyValueIterator<K, V> {
-
- private final Iterator<Map.Entry<K, V>> underlying;
-
- public TheIterator(final Iterator<Map.Entry<K, V>> iterator) {
- this.underlying = iterator;
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public K peekNextKey() {
- throw new UnsupportedOperationException("peekNextKey not supported");
- }
-
- @Override
- public boolean hasNext() {
- return underlying.hasNext();
- }
-
- @Override
- public KeyValue<K, V> next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- Map.Entry<K, V> next = underlying.next();
- return new KeyValue<>(next.getKey(), next.getValue());
- }
-
- @Override
- public void remove() {
-
- }
- }
-}
[3/5] kafka git commit: KAFKA-3452 Follow-up: Optimize ByteStore
Scenarios
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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
deleted file mode 100644
index e7c2eb3..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIteratorTest.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/**
- * 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/7ebc5da6/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
deleted file mode 100644
index 376fca8..0000000
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWindowStoreIteratorTest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/**
- * 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.metrics.Metrics;
-import org.apache.kafka.common.serialization.Serdes;
-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.streams.state.StateSerdes;
-import org.apache.kafka.test.KeyValueIteratorStub;
-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<Long, byte[]>> expectedKvPairs = new ArrayList<>();
- for (long t = 0; t < 100; t += 20) {
- final byte[] v1Bytes = String.valueOf(t).getBytes();
- 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();
- expectedKvPairs.add(KeyValue.pair(t + 10, valBytes));
- cache.put(namespace, keyBytes, new LRUCacheEntry(valBytes));
- }
-
- 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()));
- int index = 0;
- while (iterator.hasNext()) {
- final KeyValue<Long, byte[]> next = iterator.next();
- final KeyValue<Long, byte[]> expected = expectedKvPairs.get(index++);
- assertArrayEquals(expected.value, next.value);
- assertEquals(expected.key, next.key);
- }
- }
-
- @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/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedSessionStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedSessionStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedSessionStoreIteratorTest.java
new file mode 100644
index 0000000..38e577b
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedSessionStoreIteratorTest.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 MergedSortedCacheWrappedSessionStoreIteratorTest {
+
+ 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/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedWindowStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedWindowStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedWindowStoreIteratorTest.java
new file mode 100644
index 0000000..b8e9b94
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheWrappedWindowStoreIteratorTest.java
@@ -0,0 +1,89 @@
+/**
+ * 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.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;
+import org.apache.kafka.streams.state.StateSerdes;
+import org.apache.kafka.test.KeyValueIteratorStub;
+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 MergedSortedCacheWrappedWindowStoreIteratorTest {
+
+ 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<Long, byte[]>> expectedKvPairs = new ArrayList<>();
+ for (long t = 0; t < 100; t += 20) {
+ final byte[] v1Bytes = String.valueOf(t).getBytes();
+ final KeyValue<Long, byte[]> v1 = KeyValue.pair(t, v1Bytes);
+ windowStoreKvPairs.add(v1);
+ expectedKvPairs.add(KeyValue.pair(t, v1Bytes));
+ final Bytes keyBytes = WindowStoreUtils.toBinaryKey("a", t + 10, 0, stateSerdes);
+ final byte[] valBytes = String.valueOf(t + 10).getBytes();
+ expectedKvPairs.add(KeyValue.pair(t + 10, valBytes));
+ cache.put(namespace, keyBytes, new LRUCacheEntry(valBytes));
+ }
+
+ Bytes fromBytes = WindowStoreUtils.toBinaryKey("a", 0, 0, stateSerdes);
+ Bytes toBytes = WindowStoreUtils.toBinaryKey("a", 100, 0, stateSerdes);
+ final KeyValueIterator<Long, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("store", new KeyValueIteratorStub<>(windowStoreKvPairs.iterator()));
+
+ final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, fromBytes, toBytes);
+
+ 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();
+ final KeyValue<Long, byte[]> expected = expectedKvPairs.get(index++);
+ assertArrayEquals(expected.value, next.value);
+ assertEquals(expected.key, next.key);
+ }
+ }
+
+ @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()));
+ Bytes fromBytes = WindowStoreUtils.toBinaryKey("a", 0, 0, stateSerdes);
+ Bytes toBytes = WindowStoreUtils.toBinaryKey("a", 100, 0, stateSerdes);
+ final KeyValueIterator<Long, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("store", new KeyValueIteratorStub<>(windowStoreKvPairs.iterator()));
+ final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, fromBytes, toBytes);
+ 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/7ebc5da6/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 a2ce96c..92fbb2c 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
@@ -106,7 +106,7 @@ public class ReadOnlyWindowStoreStub<K, V> implements ReadOnlyWindowStore<K, V>,
@Override
public Long peekNextKey() {
- throw new UnsupportedOperationException("peekNextKey not supported in stub");
+ throw new UnsupportedOperationException("peekNextKey() not supported in " + getClass().getName());
}
@Override
[5/5] kafka git commit: KAFKA-3452 Follow-up: Optimize ByteStore
Scenarios
Posted by gu...@apache.org.
KAFKA-3452 Follow-up: Optimize ByteStore Scenarios
This is a refactoring follow-up of https://github.com/apache/kafka/pull/2166. Main refactoring changes:
1. Extract `InMemoryKeyValueStore` out of `InMemoryKeyValueStoreSupplier` and remove its duplicates in test package.
2. Add two abstract classes `AbstractKeyValueIterator` and `AbstractKeyValueStore` to collapse common functional logics.
3. Added specialized `BytesXXStore` to accommodate cases where key value types are Bytes / byte[] so that we can save calling the dummy serdes.
4. Make the key type in `ThreadCache` from byte[] to Bytes, as SessionStore / WindowStore's result serialized bytes are in the form of Bytes anyways, so that we can save unnecessary `Bytes.get()` and `Bytes.wrap(bytes)`.
Each of these should arguably be a separate PR and I apologize for the mess, this is because this branch was extracted from a rather large diff that has multiple refactoring mingled together and dguy and myself have already put lots of efforts to break it down to a few separate PRs, and this is the only left-over work. Such PR won't happen in the future.
Ping dguy enothereska mjsax for reviews
Author: Guozhang Wang <wa...@gmail.com>
Reviewers: Damian Guy, Matthias J. Sax, Jun Rao
Closes #2333 from guozhangwang/K3452-followup-state-store-refactor
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7ebc5da6
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7ebc5da6
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7ebc5da6
Branch: refs/heads/trunk
Commit: 7ebc5da606fb806bde8b7176fed0e60243e0f7f3
Parents: d95f22c
Author: Guozhang Wang <wa...@gmail.com>
Authored: Fri Feb 3 11:12:49 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Fri Feb 3 11:12:49 2017 -0800
----------------------------------------------------------------------
.../kstream/internals/AbstractStream.java | 28 +-
.../kstream/internals/SessionKeySerde.java | 23 +-
.../streams/processor/TopologyBuilder.java | 4 +-
.../processor/internals/AbstractTask.java | 1 -
.../internals/ProcessorStateManager.java | 2 -
.../kafka/streams/state/KeyValueStore.java | 5 +-
.../apache/kafka/streams/state/WindowStore.java | 1 -
.../streams/state/WindowStoreIterator.java | 2 +-
.../AbstractMergedSortedCacheStoreIterator.java | 2 +-
.../state/internals/AbstractStoreSupplier.java | 9 +-
.../state/internals/CachingKeyValueStore.java | 38 +-
.../state/internals/CachingSessionStore.java | 76 +-
.../state/internals/CachingWindowStore.java | 38 +-
.../ChangeLoggingKeyValueBytesStore.java | 13 +-
.../internals/ChangeLoggingKeyValueStore.java | 16 +-
.../ChangeLoggingSegmentedBytesStore.java | 3 +-
.../internals/CompositeReadOnlyWindowStore.java | 1 +
.../DelegatingPeekingKeyValueIterator.java | 10 +-
.../internals/InMemoryKeyValueLoggedStore.java | 40 +-
.../state/internals/InMemoryKeyValueStore.java | 187 +++
.../InMemoryKeyValueStoreSupplier.java | 165 +--
.../streams/state/internals/LRUCacheEntry.java | 17 +-
.../streams/state/internals/MemoryLRUCache.java | 2 +-
.../MergedSortedCacheKeyValueStoreIterator.java | 6 +-
.../MergedSortedCacheSessionStoreIterator.java | 2 +-
.../state/internals/MeteredKeyValueStore.java | 20 +-
.../internals/MeteredSegmentedBytesStore.java | 4 +-
.../streams/state/internals/NamedCache.java | 6 +-
.../internals/RocksDBSegmentedBytesStore.java | 14 +-
.../state/internals/RocksDBSessionStore.java | 122 +-
.../internals/RocksDBSessionStoreSupplier.java | 11 +-
.../streams/state/internals/RocksDBStore.java | 33 +-
.../state/internals/RocksDBWindowStore.java | 124 +-
.../internals/RocksDBWindowStoreSupplier.java | 2 +-
.../kafka/streams/state/internals/Segment.java | 8 +-
.../state/internals/SegmentIterator.java | 12 +-
.../state/internals/SegmentedBytesStore.java | 7 +-
.../kafka/streams/state/internals/Segments.java | 5 +-
.../internals/SerializedKeyValueIterator.java | 2 +-
.../streams/state/internals/ThreadCache.java | 40 +-
.../state/internals/WindowKeySchema.java | 59 +
.../state/internals/WindowStoreKeySchema.java | 59 -
.../state/internals/WindowStoreSupplier.java | 7 +
.../state/internals/WindowStoreUtils.java | 21 +-
.../internals/WrappedSessionStoreIterator.java | 89 ++
.../state/internals/WrappedStateStore.java | 6 +-
.../internals/WrappedWindowStoreIterator.java | 93 ++
.../kstream/internals/SessionKeySerdeTest.java | 67 +-
.../internals/ProcessorStateManagerTest.java | 1 +
.../internals/CachingKeyValueStoreTest.java | 3 +-
.../state/internals/CachingWindowStoreTest.java | 8 +-
.../ChangeLoggingKeyValueBytesStoreTest.java | 3 +-
.../ChangeLoggingKeyValueStoreTest.java | 3 +-
.../CompositeReadOnlyKeyValueStoreTest.java | 2 +-
.../DelegatingPeekingKeyValueIteratorTest.java | 4 +-
...gedSortedCacheKeyValueStoreIteratorTest.java | 45 +-
...rgedSortedCacheSessionStoreIteratorTest.java | 113 --
...ergedSortedCacheWindowStoreIteratorTest.java | 88 --
...tedCacheWrappedSessionStoreIteratorTest.java | 113 ++
...rtedCacheWrappedWindowStoreIteratorTest.java | 89 ++
.../internals/ReadOnlyWindowStoreStub.java | 2 +-
.../state/internals/RocksDBWindowStoreTest.java | 1191 +++++++-----------
.../state/internals/SegmentIteratorTest.java | 16 +-
.../state/internals/ThreadCacheTest.java | 143 ++-
.../state/internals/WindowStoreUtilsTest.java | 10 +-
.../kafka/test/InMemoryKeyValueStore.java | 152 ---
66 files changed, 1559 insertions(+), 1929 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java
index bcffce2..90a450e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java
@@ -37,7 +37,7 @@ public abstract class AbstractStream<K> {
protected final String name;
protected final Set<String> sourceNodes;
- public AbstractStream(KStreamBuilder topology, String name, Set<String> sourceNodes) {
+ AbstractStream(final KStreamBuilder topology, String name, final Set<String> sourceNodes) {
if (sourceNodes == null || sourceNodes.isEmpty()) {
throw new IllegalArgumentException("parameter <sourceNodes> must not be null or empty");
}
@@ -47,7 +47,7 @@ public abstract class AbstractStream<K> {
this.sourceNodes = sourceNodes;
}
- protected Set<String> ensureJoinableWith(AbstractStream<K> other) {
+ Set<String> ensureJoinableWith(final AbstractStream<K> other) {
Set<String> allSourceNodes = new HashSet<>();
allSourceNodes.addAll(sourceNodes);
allSourceNodes.addAll(other.sourceNodes);
@@ -57,7 +57,7 @@ public abstract class AbstractStream<K> {
return allSourceNodes;
}
- public static <T2, T1, R> ValueJoiner<T2, T1, R> reverseJoiner(final ValueJoiner<T1, T2, R> joiner) {
+ static <T2, T1, R> ValueJoiner<T2, T1, R> reverseJoiner(final ValueJoiner<T1, T2, R> joiner) {
return new ValueJoiner<T2, T1, R>() {
@Override
public R apply(T2 value2, T1 value1) {
@@ -67,27 +67,27 @@ public abstract class AbstractStream<K> {
}
@SuppressWarnings("unchecked")
- public static <T, K> StateStoreSupplier<KeyValueStore> keyValueStore(final Serde<K> keySerde,
- final Serde<T> aggValueSerde,
- final String storeName) {
+ static <T, K> StateStoreSupplier<KeyValueStore> keyValueStore(final Serde<K> keySerde,
+ final Serde<T> aggValueSerde,
+ final String storeName) {
Objects.requireNonNull(storeName, "storeName can't be null");
return storeFactory(keySerde, aggValueSerde, storeName).build();
}
@SuppressWarnings("unchecked")
- public static <W extends Window, T, K> StateStoreSupplier<WindowStore> windowedStore(final Serde<K> keySerde,
- final Serde<T> aggValSerde,
- final Windows<W> windows,
- final String storeName) {
+ static <W extends Window, T, K> StateStoreSupplier<WindowStore> windowedStore(final Serde<K> keySerde,
+ final Serde<T> aggValSerde,
+ final Windows<W> windows,
+ final String storeName) {
Objects.requireNonNull(storeName, "storeName can't be null");
return storeFactory(keySerde, aggValSerde, storeName)
.windowed(windows.size(), windows.maintainMs(), windows.segments, false)
.build();
}
- @SuppressWarnings("unchecked")
- public static <T, K> Stores.PersistentKeyValueFactory<K, T> storeFactory(final Serde<K> keySerde,
- final Serde<T> aggValueSerde,
- final String storeName) {
+
+ static <T, K> Stores.PersistentKeyValueFactory<K, T> storeFactory(final Serde<K> keySerde,
+ final Serde<T> aggValueSerde,
+ final String storeName) {
return Stores.create(storeName)
.withKeys(keySerde)
.withValues(aggValueSerde)
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
index d9a3528..16c7cdf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionKeySerde.java
@@ -112,7 +112,6 @@ public class SessionKeySerde<K> implements Serde<Windowed<K>> {
}
}
-
public static long extractEnd(final byte [] binaryKey) {
return ByteBuffer.wrap(binaryKey).getLong(binaryKey.length - 2 * TIMESTAMP_SIZE);
}
@@ -121,6 +120,13 @@ public class SessionKeySerde<K> implements Serde<Windowed<K>> {
return ByteBuffer.wrap(binaryKey).getLong(binaryKey.length - TIMESTAMP_SIZE);
}
+ public static Window extractWindow(final byte [] binaryKey) {
+ final ByteBuffer buffer = ByteBuffer.wrap(binaryKey);
+ final long start = buffer.getLong(binaryKey.length - TIMESTAMP_SIZE);
+ final long end = buffer.getLong(binaryKey.length - 2 * TIMESTAMP_SIZE);
+ return new SessionWindow(start, end);
+ }
+
public static byte[] extractKeyBytes(final byte[] binaryKey) {
final byte[] bytes = new byte[binaryKey.length - 2 * TIMESTAMP_SIZE];
System.arraycopy(binaryKey, 0, bytes, 0, bytes.length);
@@ -129,10 +135,16 @@ public class SessionKeySerde<K> implements Serde<Windowed<K>> {
public static <K> Windowed<K> from(final byte[] binaryKey, final Deserializer<K> keyDeserializer) {
final K key = extractKey(binaryKey, keyDeserializer);
+ final Window window = extractWindow(binaryKey);
+ return new Windowed<>(key, window);
+ }
+
+ public static Windowed<Bytes> fromBytes(Bytes bytesKey) {
+ final byte[] binaryKey = bytesKey.get();
final ByteBuffer buffer = ByteBuffer.wrap(binaryKey);
final long start = buffer.getLong(binaryKey.length - TIMESTAMP_SIZE);
final long end = buffer.getLong(binaryKey.length - 2 * TIMESTAMP_SIZE);
- return new Windowed<>(key, new SessionWindow(start, end));
+ return new Windowed<>(Bytes.wrap(extractKeyBytes(binaryKey)), new SessionWindow(start, end));
}
private static <K> K extractKey(final byte[] binaryKey, Deserializer<K> deserializer) {
@@ -156,11 +168,4 @@ public class SessionKeySerde<K> implements Serde<Windowed<K>> {
buf.putLong(sessionKey.window().start());
return new Bytes(buf.array());
}
-
- public static Window extractWindow(final byte [] binaryKey) {
- final ByteBuffer buffer = ByteBuffer.wrap(binaryKey);
- final long start = buffer.getLong(binaryKey.length - TIMESTAMP_SIZE);
- final long end = buffer.getLong(binaryKey.length - 2 * TIMESTAMP_SIZE);
- return new TimeWindow(start, end);
- }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
index 81f4302..b4dbf65 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
@@ -63,6 +63,8 @@ public class TopologyBuilder {
private static final Logger log = LoggerFactory.getLogger(TopologyBuilder.class);
+ private static final Pattern EMPTY_ZERO_LENGTH_PATTERN = Pattern.compile("");
+
// node factories in a topological order
private final LinkedHashMap<String, NodeFactory> nodeFactories = new LinkedHashMap<>();
@@ -113,8 +115,6 @@ public class TopologyBuilder {
private final Set<Pattern> latestResetPatterns = new HashSet<>();
- private static final Pattern EMPTY_ZERO_LENGTH_PATTERN = Pattern.compile("");
-
private final QuickUnion<String> nodeGrouper = new QuickUnion<>();
private SubscriptionUpdates subscriptionUpdates = new SubscriptionUpdates();
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
index 55418d5..2a040ba 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
@@ -71,7 +71,6 @@ public abstract class AbstractTask {
// create the processor state manager
try {
this.stateMgr = new ProcessorStateManager(id, partitions, restoreConsumer, isStandby, stateDirectory, topology.storeToChangelogTopic());
-
} catch (IOException e) {
throw new ProcessorStateException(String.format("task [%s] Error while creating the state manager", id), e);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
index ad16c77..1c786e3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
@@ -140,7 +140,6 @@ public class ProcessorStateManager implements StateManager {
// check that the underlying change log topic exist or not
String topic = storeToChangelogTopic.get(store.name());
-
if (topic == null) {
this.stores.put(store.name(), store);
return;
@@ -355,7 +354,6 @@ public class ProcessorStateManager implements StateManager {
if (stores.get(storeName).persistent() && storeToChangelogTopic.containsKey(storeName)) {
String changelogTopic = storeToChangelogTopic.get(storeName);
TopicPartition topicPartition = new TopicPartition(changelogTopic, getPartition(storeName));
-
Long offset = ackedOffsets.get(topicPartition);
if (offset != null) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
index 2a86049..cac6b75 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
@@ -37,7 +37,7 @@ public interface KeyValueStore<K, V> extends StateStore, ReadOnlyKeyValueStore<K
/**
* Update the value associated with this key
*
- * @param key They key to associate the value to
+ * @param key The key to associate the value to
* @param value The value
* @throws NullPointerException If null is used for key or value.
*/
@@ -47,7 +47,7 @@ public interface KeyValueStore<K, V> extends StateStore, ReadOnlyKeyValueStore<K
* Update the value associated with this key, unless a value
* is already associated with the key
*
- * @param key They key to associate the value to
+ * @param key The key to associate the value to
* @param value The value
* @return The old value or null if there is no such key.
* @throws NullPointerException If null is used for key or value.
@@ -70,5 +70,4 @@ public interface KeyValueStore<K, V> extends StateStore, ReadOnlyKeyValueStore<K
* @throws NullPointerException If null is used for key.
*/
V delete(K key);
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
index dee3e9b..fb70f42 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
@@ -41,5 +41,4 @@ public interface WindowStore<K, V> extends StateStore, ReadOnlyWindowStore<K, V>
* Put a key-value pair with the given timestamp into the corresponding window
*/
void put(K key, V value, long timestamp);
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
index 958b778..86aae77 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
@@ -29,7 +29,7 @@ import java.io.Closeable;
* Users need to call its {@code close} method explicitly upon completeness to release resources,
* or use try-with-resources statement (available since JDK7) for this {@link Closeable} class.
*
- * @param <E> Type of values
+ * @param <V> Type of values
*/
public interface WindowStoreIterator<V> extends KeyValueIterator<Long, V>, Closeable {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMergedSortedCacheStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMergedSortedCacheStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMergedSortedCacheStoreIterator.java
index 009dad0..bf0f33f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMergedSortedCacheStoreIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractMergedSortedCacheStoreIterator.java
@@ -154,7 +154,7 @@ abstract class AbstractMergedSortedCacheStoreIterator<K, KS, V> implements KeyVa
@Override
public void remove() {
- throw new UnsupportedOperationException("remove() is not supported");
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreSupplier.java
index 3ad44ac..cde8b36 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreSupplier.java
@@ -24,7 +24,7 @@ import org.apache.kafka.streams.processor.StateStoreSupplier;
import java.util.Map;
-public abstract class AbstractStoreSupplier<K, V, T extends StateStore> implements StateStoreSupplier<T> {
+abstract class AbstractStoreSupplier<K, V, T extends StateStore> implements StateStoreSupplier<T> {
protected final String name;
protected final Serde<K> keySerde;
protected final Serde<V> valueSerde;
@@ -32,7 +32,12 @@ public abstract class AbstractStoreSupplier<K, V, T extends StateStore> implemen
protected final boolean logged;
protected final Map<String, String> logConfig;
- public AbstractStoreSupplier(String name, Serde<K> keySerde, Serde<V> valueSerde, Time time, boolean logged, Map<String, String> logConfig) {
+ AbstractStoreSupplier(final String name,
+ final Serde<K> keySerde,
+ final Serde<V> valueSerde,
+ final Time time,
+ final boolean logged,
+ final Map<String, String> logConfig) {
this.time = time;
this.name = name;
this.valueSerde = valueSerde;
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
index 1e91b47..4f4f37b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
@@ -19,7 +19,6 @@ 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.streams.KeyValue;
-import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.kstream.internals.CacheFlushListener;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
@@ -31,7 +30,7 @@ import org.apache.kafka.streams.state.StateSerdes;
import java.util.List;
-class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K, V>, CachedStateStore<K, V> {
+class CachingKeyValueStore<K, V> extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V>, CachedStateStore<K, V> {
private final KeyValueStore<Bytes, byte[]> underlying;
private final Serde<K> keySerde;
@@ -46,6 +45,7 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying,
final Serde<K> keySerde,
final Serde<V> valueSerde) {
+ super(underlying);
this.underlying = underlying;
this.keySerde = keySerde;
this.valueSerde = valueSerde;
@@ -67,7 +67,7 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
}
@SuppressWarnings("unchecked")
- void initInternal(final ProcessorContext context) {
+ private void initInternal(final ProcessorContext context) {
this.context = (InternalProcessorContext) context;
this.serdes = new StateSerdes<>(underlying.name(),
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
@@ -83,7 +83,6 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
}
}
});
-
}
private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final InternalProcessorContext context) {
@@ -93,7 +92,8 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
if (flushListener != null) {
flushListener.apply(serdes.keyFrom(entry.key().get()),
- serdes.valueFrom(entry.newValue()), serdes.valueFrom(underlying.get(entry.key())));
+ serdes.valueFrom(entry.newValue()),
+ serdes.valueFrom(underlying.get(entry.key())));
}
underlying.put(entry.key(), entry.newValue());
@@ -139,23 +139,18 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
return get(rawKey);
}
- private void validateStoreOpen() {
- if (!isOpen()) {
- throw new InvalidStateStoreException("Store " + this.name() + " is currently closed");
- }
- }
-
private V get(final byte[] rawKey) {
- final LRUCacheEntry entry = cache.get(cacheName, rawKey);
+ final Bytes key = Bytes.wrap(rawKey);
+ final LRUCacheEntry entry = cache.get(cacheName, key);
if (entry == null) {
- final byte[] rawValue = underlying.get(Bytes.wrap(rawKey));
+ final byte[] rawValue = underlying.get(key);
if (rawValue == null) {
return null;
}
// only update the cache if this call is on the streamThread
// as we don't want other threads to trigger an eviction/flush
if (Thread.currentThread().equals(streamThread)) {
- cache.put(cacheName, rawKey, new LRUCacheEntry(rawValue));
+ cache.put(cacheName, key, new LRUCacheEntry(rawValue));
}
return serdes.valueFrom(rawValue);
}
@@ -170,9 +165,9 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
@Override
public KeyValueIterator<K, V> range(final K from, final K to) {
validateStoreOpen();
- final byte[] origFrom = serdes.rawKey(from);
- final byte[] origTo = serdes.rawKey(to);
- final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(Bytes.wrap(origFrom), Bytes.wrap(origTo));
+ final Bytes origFrom = Bytes.wrap(serdes.rawKey(from));
+ final Bytes origTo = Bytes.wrap(serdes.rawKey(to));
+ final KeyValueIterator<Bytes, byte[]> storeIterator = underlying.range(origFrom, origTo);
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, origFrom, origTo);
return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes);
}
@@ -199,8 +194,8 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
private synchronized void put(final byte[] rawKey, final V value) {
final byte[] rawValue = serdes.rawValue(value);
- cache.put(cacheName, rawKey, new LRUCacheEntry(rawValue, true, context.offset(),
- context.timestamp(), context.partition(), context.topic()));
+ cache.put(cacheName, Bytes.wrap(rawKey), new LRUCacheEntry(rawValue, true, context.offset(),
+ context.timestamp(), context.partition(), context.topic()));
}
@Override
@@ -225,9 +220,10 @@ class CachingKeyValueStore<K, V> implements WrappedStateStore, KeyValueStore<K,
public synchronized V delete(final K key) {
validateStoreOpen();
final byte[] rawKey = serdes.rawKey(key);
+ final Bytes bytesKey = Bytes.wrap(rawKey);
final V v = get(rawKey);
- cache.delete(cacheName, serdes.rawKey(key));
- underlying.delete(Bytes.wrap(rawKey));
+ cache.delete(cacheName, bytesKey);
+ underlying.delete(bytesKey);
return v;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
index 2cea915..c33e87c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
@@ -34,16 +34,16 @@ import java.util.List;
import java.util.NoSuchElementException;
-class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> {
+class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG>, CachedStateStore<Windowed<K>, AGG> {
private final SessionStore<Bytes, byte[]> bytesStore;
private final SessionKeySchema keySchema;
- private Serde<K> keySerde;
+ private final Serde<K> keySerde;
private final Serde<AGG> aggSerde;
- private InternalProcessorContext context;
private String cacheName;
- private StateSerdes<K, AGG> serdes;
private ThreadCache cache;
+ private StateSerdes<K, AGG> serdes;
+ private InternalProcessorContext context;
private CacheFlushListener<Windowed<K>, AGG> flushListener;
CachingSessionStore(final SessionStore<Bytes, byte[]> bytesStore,
@@ -56,15 +56,41 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedState
this.keySchema = new SessionKeySchema();
}
+ @SuppressWarnings("unchecked")
+ public void init(final ProcessorContext context, final StateStore root) {
+ bytesStore.init(context, root);
+ initInternal((InternalProcessorContext) context);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void initInternal(final InternalProcessorContext context) {
+ this.context = context;
+
+ this.serdes = new StateSerdes<>(bytesStore.name(),
+ keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
+ aggSerde == null ? (Serde<AGG>) context.valueSerde() : aggSerde);
+
+
+ this.cacheName = context.taskId() + "-" + bytesStore.name();
+ this.cache = this.context.getCache();
+ cache.addDirtyEntryFlushListener(cacheName, new ThreadCache.DirtyEntryFlushListener() {
+ @Override
+ public void apply(final List<ThreadCache.DirtyEntry> entries) {
+ for (ThreadCache.DirtyEntry entry : entries) {
+ putAndMaybeForward(entry, context);
+ }
+ }
+ });
+ }
+
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
validateStoreOpen();
final Bytes binarySessionId = Bytes.wrap(keySerde.serializer().serialize(this.name(), key));
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName,
- keySchema.lowerRange(binarySessionId,
- earliestSessionEndTime).get(),
- keySchema.upperRange(binarySessionId, latestSessionStartTime).get());
+ keySchema.lowerRange(binarySessionId, earliestSessionEndTime),
+ keySchema.upperRange(binarySessionId, latestSessionStartTime));
final KeyValueIterator<Windowed<Bytes>, byte[]> storeIterator = bytesStore.findSessions(binarySessionId, earliestSessionEndTime, latestSessionStartTime);
final HasNextCondition hasNextCondition = keySchema.hasNextCondition(binarySessionId,
earliestSessionEndTime,
@@ -73,18 +99,19 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedState
return new MergedSortedCacheSessionStoreIterator<>(filteredCacheIterator, storeIterator, serdes);
}
-
+ @Override
public void remove(final Windowed<K> sessionKey) {
validateStoreOpen();
put(sessionKey, null);
}
+ @Override
public void put(final Windowed<K> key, AGG value) {
validateStoreOpen();
final Bytes binaryKey = SessionKeySerde.toBinary(key, keySerde.serializer());
final LRUCacheEntry entry = new LRUCacheEntry(serdes.rawValue(value), true, context.offset(),
key.window().end(), context.partition(), context.topic());
- cache.put(cacheName, binaryKey.get(), entry);
+ cache.put(cacheName, binaryKey, entry);
}
@Override
@@ -92,34 +119,6 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedState
return findSessions(key, 0, Long.MAX_VALUE);
}
- @SuppressWarnings("unchecked")
- public void init(final ProcessorContext context, final StateStore root) {
- bytesStore.init(context, root);
- initInternal((InternalProcessorContext) context);
- }
-
- @SuppressWarnings("unchecked")
- private void initInternal(final InternalProcessorContext context) {
- this.context = context;
-
- this.serdes = new StateSerdes<>(bytesStore.name(),
- keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
- aggSerde == null ? (Serde<AGG>) context.valueSerde() : aggSerde);
-
-
- this.cacheName = context.taskId() + "-" + bytesStore.name();
- this.cache = this.context.getCache();
- cache.addDirtyEntryFlushListener(cacheName, new ThreadCache.DirtyEntryFlushListener() {
- @Override
- public void apply(final List<ThreadCache.DirtyEntry> entries) {
- for (ThreadCache.DirtyEntry entry : entries) {
- putAndMaybeForward(entry, context);
- }
- }
- });
-
- }
-
private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final InternalProcessorContext context) {
final Bytes binaryKey = entry.key();
final RecordContext current = context.recordContext();
@@ -148,7 +147,6 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedState
}
}
-
public void flush() {
cache.flush(cacheName);
bytesStore.flush();
@@ -156,8 +154,8 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractWrappedState
public void close() {
flush();
- bytesStore.close();
cache.close(cacheName);
+ bytesStore.close();
}
public void setFlushListener(CacheFlushListener<Windowed<K>, AGG> flushListener) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
index d471761..3cc8478 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
@@ -19,30 +19,31 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
-import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.CacheFlushListener;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.RecordContext;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import java.util.List;
-class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> {
+class CachingWindowStore<K, V> extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V>, CachedStateStore<Windowed<K>, V> {
private final WindowStore<Bytes, byte[]> underlying;
private final Serde<K> keySerde;
private final Serde<V> valueSerde;
- private CacheFlushListener<Windowed<K>, V> flushListener;
private final long windowSize;
+
private String name;
private ThreadCache cache;
private InternalProcessorContext context;
private StateSerdes<K, V> serdes;
+ private CacheFlushListener<Windowed<K>, V> flushListener;
CachingWindowStore(final WindowStore<Bytes, byte[]> underlying,
final Serde<K> keySerde,
@@ -63,7 +64,7 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
}
@SuppressWarnings("unchecked")
- void initInternal(final ProcessorContext context) {
+ private void initInternal(final ProcessorContext context) {
this.context = (InternalProcessorContext) context;
this.serdes = new StateSerdes<>(underlying.name(),
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
@@ -71,6 +72,7 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
this.name = context.taskId() + "-" + underlying.name();
this.cache = this.context.getCache();
+
cache.addDirtyEntryFlushListener(name, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> entries) {
@@ -79,14 +81,13 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
final long timestamp = WindowStoreUtils.timestampFromBinaryKey(binaryWindowKey);
final Windowed<K> windowedKey = new Windowed<>(WindowStoreUtils.keyFromBinaryKey(binaryWindowKey, serdes),
- new TimeWindow(timestamp, timestamp + windowSize));
+ new TimeWindow(timestamp, timestamp + windowSize));
final Bytes key = WindowStoreUtils.bytesKeyFromBinaryKey(binaryWindowKey);
maybeForward(entry, key, windowedKey, (InternalProcessorContext) context);
underlying.put(key, entry.newValue(), timestamp);
}
}
});
-
}
private void maybeForward(final ThreadCache.DirtyEntry entry,
@@ -102,7 +103,6 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
} finally {
context.setRecordContext(current);
}
-
}
}
@@ -119,8 +119,8 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
@Override
public void close() {
flush();
- underlying.close();
cache.close(name);
+ underlying.close();
}
@Override
@@ -128,30 +128,34 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractWrappedStateSto
put(key, value, context.timestamp());
}
-
@Override
public synchronized void put(final K key, final V value, final long timestamp) {
+ // since this function may not access the underlying inner store, we need to validate
+ // if store is open outside as well.
validateStoreOpen();
- final byte[] binaryKey = WindowStoreUtils.toBinaryKey(key, timestamp, 0, serdes);
+
+ final Bytes keyBytes = WindowStoreUtils.toBinaryKey(key, timestamp, 0, serdes);
final LRUCacheEntry entry = new LRUCacheEntry(serdes.rawValue(value), true, context.offset(),
timestamp, context.partition(), context.topic());
- cache.put(name, binaryKey, entry);
+ cache.put(name, keyBytes, entry);
}
@Override
public synchronized WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) {
+ // since this function may not access the underlying inner store, we need to validate
+ // if store is open outside as well.
validateStoreOpen();
- byte[] binaryFrom = WindowStoreUtils.toBinaryKey(key, timeFrom, 0, serdes);
- byte[] binaryTo = WindowStoreUtils.toBinaryKey(key, timeTo, 0, serdes);
+
+ Bytes fromBytes = WindowStoreUtils.toBinaryKey(key, timeFrom, 0, serdes);
+ Bytes toBytes = WindowStoreUtils.toBinaryKey(key, timeTo, 0, serdes);
final WindowStoreIterator<byte[]> underlyingIterator = underlying.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo);
- final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, binaryFrom, binaryTo);
+ final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, fromBytes, toBytes);
return new MergedSortedCacheWindowStoreIterator<>(cacheIterator,
underlyingIterator,
new StateSerdes<>(serdes.stateName(), Serdes.Long(), serdes.valueSerde()));
}
-
private V fetchPrevious(final Bytes key, final long timestamp) {
try (final WindowStoreIterator<byte[]> iter = underlying.fetch(key, timestamp, timestamp)) {
if (!iter.hasNext()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
index e31d04b..4f3bf48 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java
@@ -25,11 +25,11 @@ import org.apache.kafka.streams.state.KeyValueStore;
import java.util.List;
-public class ChangeLoggingKeyValueBytesStore extends WrappedStateStore.AbstractWrappedStateStore implements KeyValueStore<Bytes, byte[]> {
+public class ChangeLoggingKeyValueBytesStore extends WrappedStateStore.AbstractStateStore implements KeyValueStore<Bytes, byte[]> {
private final KeyValueStore<Bytes, byte[]> inner;
private StoreChangeLogger<Bytes, byte[]> changeLogger;
- public ChangeLoggingKeyValueBytesStore(final KeyValueStore<Bytes, byte[]> inner) {
+ ChangeLoggingKeyValueBytesStore(final KeyValueStore<Bytes, byte[]> inner) {
super(inner);
this.inner = inner;
}
@@ -40,6 +40,10 @@ public class ChangeLoggingKeyValueBytesStore extends WrappedStateStore.AbstractW
this.changeLogger = new StoreChangeLogger<>(inner.name(), context, WindowStoreUtils.INNER_SERDES);
}
+ @Override
+ public long approximateNumEntries() {
+ return inner.approximateNumEntries();
+ }
@Override
public void put(final Bytes key, final byte[] value) {
@@ -85,9 +89,4 @@ public class ChangeLoggingKeyValueBytesStore extends WrappedStateStore.AbstractW
public KeyValueIterator<Bytes, byte[]> all() {
return inner.all();
}
-
- @Override
- public long approximateNumEntries() {
- return inner.approximateNumEntries();
- }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStore.java
index cd63d1a..022803a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStore.java
@@ -28,12 +28,13 @@ import org.apache.kafka.streams.state.StateSerdes;
import java.util.ArrayList;
import java.util.List;
-class ChangeLoggingKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappedStateStore implements KeyValueStore<K, V> {
+class ChangeLoggingKeyValueStore<K, V> extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V> {
private final ChangeLoggingKeyValueBytesStore innerBytes;
private final Serde keySerde;
private final Serde valueSerde;
private StateSerdes<K, V> serdes;
+
ChangeLoggingKeyValueStore(final KeyValueStore<Bytes, byte[]> bytesStore,
final Serde keySerde,
final Serde valueSerde) {
@@ -60,6 +61,11 @@ class ChangeLoggingKeyValueStore<K, V> extends WrappedStateStore.AbstractWrapped
}
@Override
+ public long approximateNumEntries() {
+ return innerBytes.approximateNumEntries();
+ }
+
+ @Override
public void put(final K key, final V value) {
final Bytes bytesKey = Bytes.wrap(serdes.rawKey(key));
final byte[] bytesValue = serdes.rawValue(value);
@@ -106,17 +112,11 @@ class ChangeLoggingKeyValueStore<K, V> extends WrappedStateStore.AbstractWrapped
public KeyValueIterator<K, V> range(final K from, final K to) {
return new SerializedKeyValueIterator<>(innerBytes.range(Bytes.wrap(serdes.rawKey(from)),
Bytes.wrap(serdes.rawKey(to))),
- serdes);
+ serdes);
}
@Override
public KeyValueIterator<K, V> all() {
return new SerializedKeyValueIterator<>(innerBytes.all(), serdes);
}
-
- @Override
- public long approximateNumEntries() {
- return innerBytes.approximateNumEntries();
- }
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStore.java
index 21c2866..0849378 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStore.java
@@ -25,12 +25,11 @@ import org.apache.kafka.streams.state.KeyValueIterator;
* Simple wrapper around a {@link SegmentedBytesStore} to support writing
* updates to a changelog
*/
-class ChangeLoggingSegmentedBytesStore extends WrappedStateStore.AbstractWrappedStateStore implements SegmentedBytesStore {
+class ChangeLoggingSegmentedBytesStore extends WrappedStateStore.AbstractStateStore implements SegmentedBytesStore {
private final SegmentedBytesStore bytesStore;
private StoreChangeLogger<Bytes, byte[]> changeLogger;
-
ChangeLoggingSegmentedBytesStore(final SegmentedBytesStore bytesStore) {
super(bytesStore);
this.bytesStore = bytesStore;
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyWindowStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyWindowStore.java
index e0f1ec8..3ffbbb6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyWindowStore.java
@@ -56,6 +56,7 @@ public class CompositeReadOnlyWindowStore<K, V> implements ReadOnlyWindowStore<K
throw new InvalidStateStoreException("State store is not available anymore and may have been migrated to another instance; please re-discover its location from the state metadata.");
}
}
+
return new WindowStoreIterator<V>() {
@Override
public void close() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIterator.java
index f3101b1..af1fcdd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/DelegatingPeekingKeyValueIterator.java
@@ -22,10 +22,14 @@ import org.apache.kafka.streams.state.KeyValueIterator;
import java.util.NoSuchElementException;
-public class DelegatingPeekingKeyValueIterator<K, V> implements KeyValueIterator<K, V>, PeekingKeyValueIterator<K, V> {
- private final String storeName;
+/**
+ * Optimized {@link KeyValueIterator} used when the same element could be peeked multiple times.
+ */
+class DelegatingPeekingKeyValueIterator<K, V> implements KeyValueIterator<K, V>, PeekingKeyValueIterator<K, V> {
private final KeyValueIterator<K, V> underlying;
+ private final String storeName;
private KeyValue<K, V> next;
+
private volatile boolean open = true;
public DelegatingPeekingKeyValueIterator(final String storeName, final KeyValueIterator<K, V> underlying) {
@@ -76,7 +80,7 @@ public class DelegatingPeekingKeyValueIterator<K, V> implements KeyValueIterator
@Override
public void remove() {
- throw new UnsupportedOperationException("remove not supported");
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueLoggedStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueLoggedStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueLoggedStore.java
index d81f6fb..c92a1b4 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueLoggedStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueLoggedStore.java
@@ -27,40 +27,34 @@ import org.apache.kafka.streams.state.StateSerdes;
import java.util.List;
-public class InMemoryKeyValueLoggedStore<K, V> implements KeyValueStore<K, V> {
+public class InMemoryKeyValueLoggedStore<K, V> extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V> {
private final KeyValueStore<K, V> inner;
private final Serde<K> keySerde;
private final Serde<V> valueSerde;
- private final String storeName;
private StoreChangeLogger<K, V> changeLogger;
private ProcessorContext context;
- public InMemoryKeyValueLoggedStore(final String storeName, final KeyValueStore<K, V> inner, Serde<K> keySerde, Serde<V> valueSerde) {
- this.storeName = storeName;
+ InMemoryKeyValueLoggedStore(final KeyValueStore<K, V> inner, Serde<K> keySerde, Serde<V> valueSerde) {
+ super(inner);
this.inner = inner;
this.keySerde = keySerde;
this.valueSerde = valueSerde;
}
@Override
- public String name() {
- return this.storeName;
- }
-
- @Override
@SuppressWarnings("unchecked")
public void init(ProcessorContext context, StateStore root) {
this.context = context;
inner.init(context, root);
// construct the serde
- StateSerdes<K, V> serdes = new StateSerdes<>(storeName,
+ StateSerdes<K, V> serdes = new StateSerdes<>(inner.name(),
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
- this.changeLogger = new StoreChangeLogger<>(storeName, context, serdes);
+ this.changeLogger = new StoreChangeLogger<>(inner.name(), context, serdes);
// if the inner store is an LRU cache, add the eviction listener to log removed record
@@ -75,13 +69,8 @@ public class InMemoryKeyValueLoggedStore<K, V> implements KeyValueStore<K, V> {
}
@Override
- public boolean persistent() {
- return inner.persistent();
- }
-
- @Override
- public boolean isOpen() {
- return inner.isOpen();
+ public long approximateNumEntries() {
+ return inner.approximateNumEntries();
}
@Override
@@ -143,19 +132,4 @@ public class InMemoryKeyValueLoggedStore<K, V> implements KeyValueStore<K, V> {
public KeyValueIterator<K, V> all() {
return this.inner.all();
}
-
- @Override
- public long approximateNumEntries() {
- return this.inner.approximateNumEntries();
- }
-
- @Override
- public void close() {
- inner.close();
- }
-
- @Override
- public void flush() {
- this.inner.flush();
- }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
new file mode 100644
index 0000000..fe50152
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
@@ -0,0 +1,187 @@
+/**
+ * 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.streams.KeyValue;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StateSerdes;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+
+public class InMemoryKeyValueStore<K, V> implements KeyValueStore<K, V> {
+ private final String name;
+ private final Serde<K> keySerde;
+ private final Serde<V> valueSerde;
+ private final NavigableMap<K, V> map;
+ private volatile boolean open = false;
+
+ private StateSerdes<K, V> serdes;
+
+ InMemoryKeyValueStore(final String name, final Serde<K> keySerde, final Serde<V> valueSerde) {
+ this.name = name;
+ this.keySerde = keySerde;
+ this.valueSerde = valueSerde;
+
+ // TODO: when we have serde associated with class types, we can
+ // improve this situation by passing the comparator here.
+ this.map = new TreeMap<>();
+ }
+
+ public KeyValueStore<K, V> enableLogging() {
+ return new InMemoryKeyValueLoggedStore<>(this, keySerde, valueSerde);
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void init(ProcessorContext context, StateStore root) {
+ // construct the serde
+ this.serdes = new StateSerdes<>(name,
+ keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
+ valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
+
+ if (root != null) {
+ // register the store
+ context.register(root, true, new StateRestoreCallback() {
+ @Override
+ public void restore(byte[] key, byte[] value) {
+ // check value for null, to avoid deserialization error.
+ if (value == null) {
+ put(serdes.keyFrom(key), null);
+ } else {
+ put(serdes.keyFrom(key), serdes.valueFrom(value));
+ }
+ }
+ });
+ }
+
+ this.open = true;
+ }
+
+ @Override
+ public boolean persistent() {
+ return false;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return this.open;
+ }
+
+ @Override
+ public synchronized V get(K key) {
+ return this.map.get(key);
+ }
+
+ @Override
+ public synchronized void put(K key, V value) {
+ this.map.put(key, value);
+ }
+
+ @Override
+ public synchronized V putIfAbsent(K key, V value) {
+ V originalValue = get(key);
+ if (originalValue == null) {
+ put(key, value);
+ }
+ return originalValue;
+ }
+
+ @Override
+ public synchronized void putAll(List<KeyValue<K, V>> entries) {
+ for (KeyValue<K, V> entry : entries)
+ put(entry.key, entry.value);
+ }
+
+ @Override
+ public synchronized V delete(K key) {
+ return this.map.remove(key);
+ }
+
+ @Override
+ public synchronized KeyValueIterator<K, V> range(K from, K to) {
+ return new DelegatingPeekingKeyValueIterator<>(name, new InMemoryKeyValueIterator<>(this.map.subMap(from, true, to, true).entrySet().iterator()));
+ }
+
+ @Override
+ public synchronized KeyValueIterator<K, V> all() {
+ final TreeMap<K, V> copy = new TreeMap<>(this.map);
+ return new DelegatingPeekingKeyValueIterator<>(name, new InMemoryKeyValueIterator<>(copy.entrySet().iterator()));
+ }
+
+ @Override
+ public long approximateNumEntries() {
+ return this.map.size();
+ }
+
+ @Override
+ public void flush() {
+ // do-nothing since it is in-memory
+ }
+
+ @Override
+ public void close() {
+ this.map.clear();
+ this.open = false;
+ }
+
+ private static class InMemoryKeyValueIterator<K, V> implements KeyValueIterator<K, V> {
+ private final Iterator<Map.Entry<K, V>> iter;
+
+ private InMemoryKeyValueIterator(Iterator<Map.Entry<K, V>> iter) {
+ this.iter = iter;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public KeyValue<K, V> next() {
+ Map.Entry<K, V> entry = iter.next();
+ return new KeyValue<>(entry.getKey(), entry.getValue());
+ }
+
+ @Override
+ public void remove() {
+ iter.remove();
+ }
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+
+ @Override
+ public K peekNextKey() {
+ throw new UnsupportedOperationException("peekNextKey() not supported in " + getClass().getName());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreSupplier.java
index e00f8ab..4be2c4f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreSupplier.java
@@ -19,19 +19,9 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.StateRestoreCallback;
-import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
-import org.apache.kafka.streams.state.StateSerdes;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.NavigableMap;
-import java.util.TreeMap;
/**
* An in-memory key-value store based on a TreeMap.
@@ -47,7 +37,6 @@ import java.util.TreeMap;
*/
public class InMemoryKeyValueStoreSupplier<K, V> extends AbstractStoreSupplier<K, V, KeyValueStore> {
-
public InMemoryKeyValueStoreSupplier(String name, Serde<K> keySerde, Serde<V> valueSerde, boolean logged, Map<String, String> logConfig) {
this(name, keySerde, valueSerde, null, logged, logConfig);
}
@@ -57,160 +46,8 @@ public class InMemoryKeyValueStoreSupplier<K, V> extends AbstractStoreSupplier<K
}
public KeyValueStore get() {
- MemoryStore<K, V> store = new MemoryStore<>(name, keySerde, valueSerde);
+ InMemoryKeyValueStore<K, V> store = new InMemoryKeyValueStore<>(name, keySerde, valueSerde);
return new MeteredKeyValueStore<>(logged ? store.enableLogging() : store, "in-memory-state", time);
}
-
- private static class MemoryStore<K, V> implements KeyValueStore<K, V> {
- private final String name;
- private final Serde<K> keySerde;
- private final Serde<V> valueSerde;
- private final NavigableMap<K, V> map;
- private volatile boolean open = false;
-
- private StateSerdes<K, V> serdes;
-
- public MemoryStore(String name, Serde<K> keySerde, Serde<V> valueSerde) {
- this.name = name;
- this.keySerde = keySerde;
- this.valueSerde = valueSerde;
-
- // TODO: when we have serde associated with class types, we can
- // improve this situation by passing the comparator here.
- this.map = new TreeMap<>();
- }
-
- public KeyValueStore<K, V> enableLogging() {
- return new InMemoryKeyValueLoggedStore<>(this.name, this, keySerde, valueSerde);
- }
-
- @Override
- public String name() {
- return this.name;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public void init(ProcessorContext context, StateStore root) {
- // construct the serde
- this.serdes = new StateSerdes<>(name,
- keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
- valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
-
- // register the store
- context.register(root, true, new StateRestoreCallback() {
- @Override
- public void restore(byte[] key, byte[] value) {
- // check value for null, to avoid deserialization error.
- if (value == null) {
- put(serdes.keyFrom(key), null);
- } else {
- put(serdes.keyFrom(key), serdes.valueFrom(value));
- }
- }
- });
- this.open = true;
- }
-
- @Override
- public boolean persistent() {
- return false;
- }
-
- @Override
- public boolean isOpen() {
- return this.open;
- }
-
- @Override
- public synchronized V get(K key) {
- return this.map.get(key);
- }
-
- @Override
- public synchronized void put(K key, V value) {
- this.map.put(key, value);
- }
-
- @Override
- public synchronized V putIfAbsent(K key, V value) {
- V originalValue = get(key);
- if (originalValue == null) {
- put(key, value);
- }
- return originalValue;
- }
-
- @Override
- public synchronized void putAll(List<KeyValue<K, V>> entries) {
- for (KeyValue<K, V> entry : entries)
- put(entry.key, entry.value);
- }
-
- @Override
- public synchronized V delete(K key) {
- return this.map.remove(key);
- }
-
- @Override
- public synchronized KeyValueIterator<K, V> range(K from, K to) {
- return new DelegatingPeekingKeyValueIterator<>(name, new MemoryStoreIterator<>(this.map.subMap(from, true, to, false).entrySet().iterator()));
- }
-
- @Override
- public synchronized KeyValueIterator<K, V> all() {
- final TreeMap<K, V> copy = new TreeMap<>(this.map);
- return new DelegatingPeekingKeyValueIterator<>(name, new MemoryStoreIterator<>(copy.entrySet().iterator()));
- }
-
- @Override
- public long approximateNumEntries() {
- return this.map.size();
- }
-
- @Override
- public void flush() {
- // do-nothing since it is in-memory
- }
-
- @Override
- public void close() {
- this.open = false;
- }
-
- private static class MemoryStoreIterator<K, V> implements KeyValueIterator<K, V> {
- private final Iterator<Map.Entry<K, V>> iter;
-
- public MemoryStoreIterator(Iterator<Map.Entry<K, V>> iter) {
- this.iter = iter;
- }
-
- @Override
- public boolean hasNext() {
- return iter.hasNext();
- }
-
- @Override
- public KeyValue<K, V> next() {
- Map.Entry<K, V> entry = iter.next();
- return new KeyValue<>(entry.getKey(), entry.getValue());
- }
-
- @Override
- public void remove() {
- iter.remove();
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public K peekNextKey() {
- throw new UnsupportedOperationException("peekNextKey not supported on MemoryStoreIterator");
- }
-
- }
- }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
index b7e47e2..e20f409 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
@@ -25,12 +25,12 @@ class LRUCacheEntry implements RecordContext {
public final byte[] value;
private final long offset;
- private final long timestamp;
private final String topic;
- boolean isDirty;
private final int partition;
- private long sizeBytes = 0;
+ private final long timestamp;
+ private long sizeBytes;
+ private boolean isDirty;
LRUCacheEntry(final byte[] value) {
this(value, false, -1, -1, -1, "");
@@ -51,13 +51,6 @@ class LRUCacheEntry implements RecordContext {
8 + // offset
4 + // partition
(topic == null ? 0 : topic.length());
-
- }
-
-
-
- void markClean() {
- isDirty = false;
}
@Override
@@ -80,6 +73,10 @@ class LRUCacheEntry implements RecordContext {
return partition;
}
+ void markClean() {
+ isDirty = false;
+ }
+
boolean isDirty() {
return isDirty;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java
index 4e1f40e..c1eb689 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java
@@ -85,7 +85,7 @@ public class MemoryLRUCache<K, V> implements KeyValueStore<K, V> {
}
public KeyValueStore<K, V> enableLogging() {
- return new InMemoryKeyValueLoggedStore<>(this.name, this, keySerde, valueSerde);
+ return new InMemoryKeyValueLoggedStore<>(this, keySerde, valueSerde);
}
public MemoryLRUCache<K, V> whenEldestRemoved(EldestEntryRemovalListener<K, V> listener) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIterator.java
index b860e16..91673da 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIterator.java
@@ -36,7 +36,7 @@ class MergedSortedCacheKeyValueStoreIterator<K, V> extends AbstractMergedSortedC
}
@Override
- public KeyValue<K, V> deserializeStorePair(KeyValue<Bytes, byte[]> pair) {
+ public KeyValue<K, V> deserializeStorePair(final KeyValue<Bytes, byte[]> pair) {
return KeyValue.pair(serdes.keyFrom(pair.key.get()), serdes.valueFrom(pair.value));
}
@@ -46,12 +46,12 @@ class MergedSortedCacheKeyValueStoreIterator<K, V> extends AbstractMergedSortedC
}
@Override
- public K deserializeStoreKey(Bytes key) {
+ public K deserializeStoreKey(final Bytes key) {
return serdes.keyFrom(key.get());
}
@Override
- public int compare(Bytes cacheKey, Bytes storeKey) {
+ public int compare(final Bytes cacheKey, final Bytes storeKey) {
return cacheKey.compareTo(storeKey);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIterator.java
index db64621..db7e898 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MergedSortedCacheSessionStoreIterator.java
@@ -59,7 +59,7 @@ class MergedSortedCacheSessionStoreIterator<K, AGG> extends AbstractMergedSorted
@Override
public Windowed<K> deserializeStoreKey(Windowed<Bytes> key) {
final K originalKey = rawSerdes.keyFrom(key.key().get());
- return new Windowed<K>(originalKey, key.window());
+ return new Windowed<>(originalKey, key.window());
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
index dc6b7a5..0a72759 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
@@ -29,16 +29,16 @@ import org.apache.kafka.streams.state.KeyValueStore;
import java.util.List;
/**
- * Metered KeyValueStore wrapper is used for recording operation metrics, and hence its
+ * Metered {@link KeyValueStore} wrapper is used for recording operation metrics, and hence its
* inner KeyValueStore implementation do not need to provide its own metrics collecting functionality.
*
* @param <K>
* @param <V>
*/
-public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappedStateStore implements KeyValueStore<K, V> {
+public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractStateStore implements KeyValueStore<K, V> {
- protected final KeyValueStore<K, V> inner;
- protected final String metricScope;
+ private final KeyValueStore<K, V> inner;
+ private final String metricScope;
protected final Time time;
private Sensor putTime;
@@ -132,6 +132,11 @@ public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappe
}
@Override
+ public long approximateNumEntries() {
+ return inner.approximateNumEntries();
+ }
+
+ @Override
public V get(K key) {
this.key = key;
metrics.measureLatencyNs(time, getDelegate, this.getTime);
@@ -177,11 +182,6 @@ public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappe
}
@Override
- public long approximateNumEntries() {
- return this.inner.approximateNumEntries();
- }
-
- @Override
public void flush() {
metrics.measureLatencyNs(time, flushDelegate, this.flushTime);
}
@@ -192,7 +192,7 @@ public class MeteredKeyValueStore<K, V> extends WrappedStateStore.AbstractWrappe
private final Sensor sensor;
private final long startNs;
- public MeteredKeyValueIterator(KeyValueIterator<K1, V1> iter, Sensor sensor) {
+ MeteredKeyValueIterator(KeyValueIterator<K1, V1> iter, Sensor sensor) {
this.iter = iter;
this.sensor = sensor;
this.startNs = time.nanoseconds();
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
index 4d3771e..3bc1bf6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSegmentedBytesStore.java
@@ -27,7 +27,7 @@ import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.KeyValueIterator;
-class MeteredSegmentedBytesStore extends WrappedStateStore.AbstractWrappedStateStore implements SegmentedBytesStore {
+class MeteredSegmentedBytesStore extends WrappedStateStore.AbstractStateStore implements SegmentedBytesStore {
private final SegmentedBytesStore inner;
private final String metricScope;
@@ -154,7 +154,5 @@ class MeteredSegmentedBytesStore extends WrappedStateStore.AbstractWrappedStateS
public Bytes peekNextKey() {
return iter.peekNextKey();
}
-
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
index 229ccec..931ab17 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java
@@ -152,7 +152,7 @@ class NamedCache {
synchronized void put(final Bytes key, final LRUCacheEntry value) {
- if (!value.isDirty && dirtyKeys.contains(key)) {
+ if (!value.isDirty() && dirtyKeys.contains(key)) {
throw new IllegalStateException(String.format("Attempting to put a clean entry for key [%s] " +
"into NamedCache [%s] when it already contains " +
"a dirty entry for the same key",
@@ -308,10 +308,6 @@ class NamedCache {
return tail;
}
- synchronized long dirtySize() {
- return dirtyKeys.size();
- }
-
synchronized void close() {
head = tail = null;
listener = null;
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStore.java
index 31956ba..ecd4088 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSegmentedBytesStore.java
@@ -32,7 +32,6 @@ class RocksDBSegmentedBytesStore implements SegmentedBytesStore {
private ProcessorContext context;
private volatile boolean open;
-
RocksDBSegmentedBytesStore(final String name,
final long retention,
final int numSegments,
@@ -42,19 +41,16 @@ class RocksDBSegmentedBytesStore implements SegmentedBytesStore {
this.segments = new Segments(name, retention, numSegments);
}
-
@Override
- @SuppressWarnings("unchecked")
public KeyValueIterator<Bytes, byte[]> fetch(final Bytes key, final long from, final long to) {
final List<Segment> searchSpace = keySchema.segmentsToSearch(segments, from, to);
final Bytes binaryFrom = keySchema.lowerRange(key, from);
final Bytes binaryTo = keySchema.upperRange(key, to);
- return new SegmentIterator(
- searchSpace.iterator(),
- keySchema.hasNextCondition(key, from, to),
- binaryFrom, binaryTo);
+ return new SegmentIterator(searchSpace.iterator(),
+ keySchema.hasNextCondition(key, from, to),
+ binaryFrom, binaryTo);
}
@Override
@@ -75,7 +71,6 @@ class RocksDBSegmentedBytesStore implements SegmentedBytesStore {
}
}
-
@Override
public byte[] get(final Bytes key) {
final Segment segment = segments.getSegmentForTimestamp(keySchema.segmentTimestamp(key));
@@ -91,7 +86,6 @@ class RocksDBSegmentedBytesStore implements SegmentedBytesStore {
}
@Override
- @SuppressWarnings("unchecked")
public void init(ProcessorContext context, StateStore root) {
this.context = context;
@@ -129,6 +123,4 @@ class RocksDBSegmentedBytesStore implements SegmentedBytesStore {
public boolean isOpen() {
return open;
}
-
-
}
[2/5] kafka git commit: KAFKA-3452 Follow-up: Optimize ByteStore
Scenarios
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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 d606e84..6286dde 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
@@ -20,7 +20,6 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
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.Utils;
@@ -36,12 +35,10 @@ import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.apache.kafka.test.MockProcessorContext;
import org.apache.kafka.test.TestUtils;
-import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
-import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -59,30 +56,42 @@ import static org.junit.Assert.fail;
public class RocksDBWindowStoreTest {
- private final Serde<byte[]> byteArraySerde = Serdes.ByteArray();
- private final String windowName = "window";
+ private static final long DEFAULT_CACHE_SIZE_BYTES = 1024 * 1024L;
+
private final int numSegments = 3;
+ private final long windowSize = 3;
+ private final String windowName = "window";
private final long segmentSize = Segments.MIN_SEGMENT_INTERVAL;
private final long retentionPeriod = segmentSize * (numSegments - 1);
- private final long windowSize = 3;
- private final Serde<Integer> intSerde = Serdes.Integer();
- private final Serde<String> stringSerde = Serdes.String();
- private final StateSerdes<Integer, String> serdes = new StateSerdes<>("", intSerde, stringSerde);
- private static final long DEFAULT_CACHE_SIZE_BYTES = 1 * 1024 * 1024L;
- private ThreadCache cache;
private final Segments segments = new Segments(windowName, retentionPeriod, numSegments);
+ private final StateSerdes<Integer, String> serdes = new StateSerdes<>("", Serdes.Integer(), Serdes.String());
+
+ private final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
+ private final ThreadCache cache = new ThreadCache("TestCache", DEFAULT_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
+
+ private final Producer<byte[], byte[]> producer = new MockProducer<>(true, Serdes.ByteArray().serializer(), Serdes.ByteArray().serializer());
+ private final RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTestTask") {
+ @Override
+ public <K1, V1> void send(final String topic,
+ K1 key,
+ V1 value,
+ Integer partition,
+ Long timestamp,
+ Serializer<K1> keySerializer,
+ Serializer<V1> valueSerializer) {
+ changeLog.add(new KeyValue<>(
+ keySerializer.serialize(topic, key),
+ valueSerializer.serialize(topic, value))
+ );
+ }
+ };
- @Before
- public void setUp() {
- cache = new ThreadCache("testCache", DEFAULT_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
- }
-
-
-
+ private final File baseDir = TestUtils.tempDirectory("test");
+ private final MockProcessorContext context = new MockProcessorContext(baseDir, Serdes.ByteArray(), Serdes.ByteArray(), recordCollector, cache);
@SuppressWarnings("unchecked")
- protected <K, V> WindowStore<K, V> createWindowStore(ProcessorContext context, final boolean enableCaching, final boolean retainDuplicates) {
- final RocksDBWindowStoreSupplier supplier = new RocksDBWindowStoreSupplier<>(windowName, retentionPeriod, numSegments, retainDuplicates, intSerde, stringSerde, windowSize, true, Collections.<String, String>emptyMap(), enableCaching);
+ private <K, V> WindowStore<K, V> createWindowStore(ProcessorContext context, final boolean enableCaching, final boolean retainDuplicates) {
+ final RocksDBWindowStoreSupplier supplier = new RocksDBWindowStoreSupplier<>(windowName, retentionPeriod, numSegments, retainDuplicates, Serdes.Integer(), Serdes.String(), windowSize, true, Collections.<String, String>emptyMap(), enableCaching);
final WindowStore<K, V> store = (WindowStore<K, V>) supplier.get();
store.init(context, store);
return store;
@@ -90,25 +99,6 @@ public class RocksDBWindowStoreTest {
@Test
public void shouldOnlyIterateOpenSegments() throws Exception {
- final File baseDir = TestUtils.tempDirectory();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-ShouldOnlyIterateOpenSegments") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
final WindowStore<Integer, String> windowStore = createWindowStore(context, false, true);
long currentTime = 0;
context.setRecordContext(createRecordContext(currentTime));
@@ -141,810 +131,491 @@ public class RocksDBWindowStoreTest {
@Test
public void testPutAndFetch() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetch") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- long startTime = segmentSize - 4L;
-
- putFirstBatch(store, startTime, context);
-
- assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L + windowSize)));
- assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L + windowSize)));
-
- putSecondBatch(store, startTime, context);
-
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L - windowSize, startTime - 2L + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L + windowSize)));
- assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L + windowSize)));
- assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L + windowSize)));
- assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L + windowSize)));
- assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L + windowSize)));
- assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L + windowSize)));
- assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L + windowSize)));
-
- // Flush the store and verify all current entries were properly flushed ...
- store.flush();
-
- Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
-
- assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
- assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
- assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
- assertNull(entriesByKey.get(3));
- assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
- assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
- assertNull(entriesByKey.get(6));
-
- } finally {
- store.close();
- }
+ long startTime = segmentSize - 4L;
+
+ putFirstBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L + windowSize)));
+ assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L + windowSize)));
+
+ putSecondBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L - windowSize, startTime - 2L + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L + windowSize)));
+ assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L + windowSize)));
+ assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L + windowSize)));
+ assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L + windowSize)));
+ assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L + windowSize)));
+ assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L + windowSize)));
+ assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L + windowSize)));
+
+ // Flush the store and verify all current entries were properly flushed ...
+ store.flush();
+
+ Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
+
+ assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
+ assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
+ assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
+ assertNull(entriesByKey.get(3));
+ assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
+ assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
+ assertNull(entriesByKey.get(6));
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
@Test
public void testPutAndFetchBefore() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetchBefore") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- long startTime = segmentSize - 4L;
-
- putFirstBatch(store, startTime, context);
-
- assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L)));
- assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L)));
-
- putSecondBatch(store, startTime, context);
-
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 0L - windowSize, startTime + 0L)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L)));
- assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L)));
- assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L)));
- assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L)));
- assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L)));
- assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L)));
- assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L)));
- assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L)));
- assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 13L - windowSize, startTime + 13L)));
-
- // Flush the store and verify all current entries were properly flushed ...
- store.flush();
-
- Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
-
- assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
- assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
- assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
- assertNull(entriesByKey.get(3));
- assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
- assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
- assertNull(entriesByKey.get(6));
-
- } finally {
- store.close();
- }
+ long startTime = segmentSize - 4L;
+
+ putFirstBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L)));
+ assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L)));
+
+ putSecondBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 0L - windowSize, startTime + 0L)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L)));
+ assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L)));
+ assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L)));
+ assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L)));
+ assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L)));
+ assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L)));
+ assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L)));
+ assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 13L - windowSize, startTime + 13L)));
+
+ // Flush the store and verify all current entries were properly flushed ...
+ store.flush();
+
+ Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
+
+ assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
+ assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
+ assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
+ assertNull(entriesByKey.get(3));
+ assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
+ assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
+ assertNull(entriesByKey.get(6));
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
- private void putSecondBatch(final WindowStore<Integer, String> store, final long startTime, MockProcessorContext context) {
- context.setRecordContext(createRecordContext(startTime + 3L));
- store.put(2, "two+1");
- context.setRecordContext(createRecordContext(startTime + 4L));
- store.put(2, "two+2");
- context.setRecordContext(createRecordContext(startTime + 5L));
- store.put(2, "two+3");
- context.setRecordContext(createRecordContext(startTime + 6L));
- store.put(2, "two+4");
- context.setRecordContext(createRecordContext(startTime + 7L));
- store.put(2, "two+5");
- context.setRecordContext(createRecordContext(startTime + 8L));
- store.put(2, "two+6");
- }
-
@Test
public void testPutAndFetchAfter() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetchAfter") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- long startTime = segmentSize - 4L;
-
- putFirstBatch(store, startTime, context);
-
- assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L, startTime + 0L + windowSize)));
- assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L, startTime + 1L + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L, startTime + 3L + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L, startTime + 4L + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L, startTime + 5L + windowSize)));
-
- putSecondBatch(store, startTime, context);
-
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L, startTime - 2L + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L, startTime - 1L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime, startTime + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L, startTime + 1L + windowSize)));
- assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize)));
- assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L, startTime + 3L + windowSize)));
- assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L, startTime + 4L + windowSize)));
- assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L, startTime + 5L + windowSize)));
- assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L, startTime + 6L + windowSize)));
- assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 7L, startTime + 7L + windowSize)));
- assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 8L, startTime + 8L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 9L, startTime + 9L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 10L, startTime + 10L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 11L, startTime + 11L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L, startTime + 12L + windowSize)));
-
- // Flush the store and verify all current entries were properly flushed ...
- store.flush();
-
- Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
-
- assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
- assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
- assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
- assertNull(entriesByKey.get(3));
- assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
- assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
- assertNull(entriesByKey.get(6));
-
- } finally {
- store.close();
- }
+ long startTime = segmentSize - 4L;
+
+ putFirstBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L, startTime + 0L + windowSize)));
+ assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L, startTime + 1L + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L, startTime + 3L + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L, startTime + 4L + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L, startTime + 5L + windowSize)));
+
+ putSecondBatch(store, startTime, context);
+
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L, startTime - 2L + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L, startTime - 1L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime, startTime + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L, startTime + 1L + windowSize)));
+ assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize)));
+ assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L, startTime + 3L + windowSize)));
+ assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L, startTime + 4L + windowSize)));
+ assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L, startTime + 5L + windowSize)));
+ assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L, startTime + 6L + windowSize)));
+ assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 7L, startTime + 7L + windowSize)));
+ assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 8L, startTime + 8L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 9L, startTime + 9L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 10L, startTime + 10L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 11L, startTime + 11L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L, startTime + 12L + windowSize)));
+
+ // Flush the store and verify all current entries were properly flushed ...
+ store.flush();
+
+ Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
+
+ assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0));
+ assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1));
+ assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2));
+ assertNull(entriesByKey.get(3));
+ assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4));
+ assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5));
+ assertNull(entriesByKey.get(6));
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
- private void putFirstBatch(final WindowStore<Integer, String> store, final long startTime, final MockProcessorContext context) {
- context.setRecordContext(createRecordContext(startTime));
- store.put(0, "zero");
- context.setRecordContext(createRecordContext(startTime + 1L));
- store.put(1, "one");
- context.setRecordContext(createRecordContext(startTime + 2L));
- store.put(2, "two");
- context.setRecordContext(createRecordContext(startTime + 4L));
- store.put(4, "four");
- context.setRecordContext(createRecordContext(startTime + 5L));
- store.put(5, "five");
- }
-
@Test
public void testPutSameKeyTimestamp() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutSameKeyTimestamp") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
+ long startTime = segmentSize - 4L;
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- long startTime = segmentSize - 4L;
+ context.setRecordContext(createRecordContext(startTime));
+ store.put(0, "zero");
- context.setRecordContext(createRecordContext(startTime));
- store.put(0, "zero");
+ assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ store.put(0, "zero");
+ store.put(0, "zero+");
+ store.put(0, "zero++");
- store.put(0, "zero");
- store.put(0, "zero+");
- store.put(0, "zero++");
+ assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 1L - windowSize, startTime + 1L + windowSize)));
+ assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 2L - windowSize, startTime + 2L + windowSize)));
+ assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 3L - windowSize, startTime + 3L + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(0, startTime + 4L - windowSize, startTime + 4L + windowSize)));
- assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 1L - windowSize, startTime + 1L + windowSize)));
- assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 2L - windowSize, startTime + 2L + windowSize)));
- assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 3L - windowSize, startTime + 3L + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(0, startTime + 4L - windowSize, startTime + 4L + windowSize)));
+ // Flush the store and verify all current entries were properly flushed ...
+ store.flush();
- // Flush the store and verify all current entries were properly flushed ...
- store.flush();
+ Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
- Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime);
-
- assertEquals(Utils.mkSet("zero@0", "zero@0", "zero+@0", "zero++@0"), entriesByKey.get(0));
-
- } finally {
- store.close();
- }
+ assertEquals(Utils.mkSet("zero@0", "zero@0", "zero+@0", "zero++@0"), entriesByKey.get(0));
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
@Test
public void testCachingEnabled() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
- try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "anyTaskID") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, true, false);
- assertTrue(store instanceof CachedStateStore);
- } finally {
- Utils.delete(baseDir);
- }
+ WindowStore<Integer, String> store = createWindowStore(context, true, false);
+ assertTrue(store instanceof CachedStateStore);
}
@Test
public void testRolling() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
+
try {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRolling") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- // to validate segments
- final Segments segments = new Segments(windowName, retentionPeriod, numSegments);
- long startTime = segmentSize * 2;
- long incr = segmentSize / 2;
- context.setRecordContext(createRecordContext(startTime));
- store.put(0, "zero");
- assertEquals(Utils.mkSet(segments.segmentName(2)), segmentDirs(baseDir));
-
- context.setRecordContext(createRecordContext(startTime + incr));
- store.put(1, "one");
- assertEquals(Utils.mkSet(segments.segmentName(2)), segmentDirs(baseDir));
-
- context.setRecordContext(createRecordContext(startTime + incr * 2));
- store.put(2, "two");
- assertEquals(Utils.mkSet(segments.segmentName(2),
- segments.segmentName(3)), segmentDirs(baseDir));
-
-
- context.setRecordContext(createRecordContext(startTime + incr * 4));
- store.put(4, "four");
- assertEquals(Utils.mkSet(segments.segmentName(2),
- segments.segmentName(3),
- segments.segmentName(4)), segmentDirs(baseDir));
-
-
- context.setRecordContext(createRecordContext(startTime + incr * 5));
- store.put(5, "five");
- assertEquals(Utils.mkSet(segments.segmentName(2),
- segments.segmentName(3),
- segments.segmentName(4)), segmentDirs(baseDir));
-
- assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
-
- context.setRecordContext(createRecordContext(startTime + incr * 6));
- store.put(6, "six");
- assertEquals(Utils.mkSet(segments.segmentName(3),
- segments.segmentName(4),
- segments.segmentName(5)), segmentDirs(baseDir));
-
-
- assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
- assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
-
-
- context.setRecordContext(createRecordContext(startTime + incr * 7));
- store.put(7, "seven");
- assertEquals(Utils.mkSet(segments.segmentName(3),
- segments.segmentName(4),
- segments.segmentName(5)), segmentDirs(baseDir));
-
- assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
- assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
- assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
- assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
-
- context.setRecordContext(createRecordContext(startTime + incr * 8));
- store.put(8, "eight");
- assertEquals(Utils.mkSet(segments.segmentName(4),
- segments.segmentName(5),
- segments.segmentName(6)), segmentDirs(baseDir));
-
-
- assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
- assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
- assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
- assertEquals(Utils.mkList("eight"), toList(store.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize)));
-
- // check segment directories
- store.flush();
- assertEquals(Utils.mkSet(segments.segmentName(4),
- segments.segmentName(5),
- segments.segmentName(6)), segmentDirs(baseDir));
-
- } finally {
- store.close();
- }
+ // to validate segments
+ final Segments segments = new Segments(windowName, retentionPeriod, numSegments);
+ long startTime = segmentSize * 2;
+ long incr = segmentSize / 2;
+ context.setRecordContext(createRecordContext(startTime));
+ store.put(0, "zero");
+ assertEquals(Utils.mkSet(segments.segmentName(2)), segmentDirs(baseDir));
+
+ context.setRecordContext(createRecordContext(startTime + incr));
+ store.put(1, "one");
+ assertEquals(Utils.mkSet(segments.segmentName(2)), segmentDirs(baseDir));
+
+ context.setRecordContext(createRecordContext(startTime + incr * 2));
+ store.put(2, "two");
+ assertEquals(Utils.mkSet(segments.segmentName(2),
+ segments.segmentName(3)), segmentDirs(baseDir));
+
+ context.setRecordContext(createRecordContext(startTime + incr * 4));
+ store.put(4, "four");
+ assertEquals(Utils.mkSet(segments.segmentName(2),
+ segments.segmentName(3),
+ segments.segmentName(4)), segmentDirs(baseDir));
+
+
+ context.setRecordContext(createRecordContext(startTime + incr * 5));
+ store.put(5, "five");
+ assertEquals(Utils.mkSet(segments.segmentName(2),
+ segments.segmentName(3),
+ segments.segmentName(4)), segmentDirs(baseDir));
+
+ assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+
+ context.setRecordContext(createRecordContext(startTime + incr * 6));
+ store.put(6, "six");
+ assertEquals(Utils.mkSet(segments.segmentName(3),
+ segments.segmentName(4),
+ segments.segmentName(5)), segmentDirs(baseDir));
+
+
+ assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+ assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
+
+
+ context.setRecordContext(createRecordContext(startTime + incr * 7));
+ store.put(7, "seven");
+ assertEquals(Utils.mkSet(segments.segmentName(3),
+ segments.segmentName(4),
+ segments.segmentName(5)), segmentDirs(baseDir));
+
+ assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+ assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
+ assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
+
+ context.setRecordContext(createRecordContext(startTime + incr * 8));
+ store.put(8, "eight");
+ assertEquals(Utils.mkSet(segments.segmentName(4),
+ segments.segmentName(5),
+ segments.segmentName(6)), segmentDirs(baseDir));
+
+
+ assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+ assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
+ assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
+ assertEquals(Utils.mkList("eight"), toList(store.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize)));
+
+ // check segment directories
+ store.flush();
+ assertEquals(Utils.mkSet(segments.segmentName(4),
+ segments.segmentName(5),
+ segments.segmentName(6)), segmentDirs(baseDir));
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
@Test
public void testRestore() throws IOException {
- final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>();
long startTime = segmentSize * 2;
long incr = segmentSize / 2;
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRestore") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
- context.setRecordContext(createRecordContext(startTime));
- store.put(0, "zero");
- context.setRecordContext(createRecordContext(startTime + incr));
- store.put(1, "one");
- context.setRecordContext(createRecordContext(startTime + incr * 2));
- store.put(2, "two");
- context.setRecordContext(createRecordContext(startTime + incr * 3));
- store.put(3, "three");
- context.setRecordContext(createRecordContext(startTime + incr * 4));
- store.put(4, "four");
- context.setRecordContext(createRecordContext(startTime + incr * 5));
- store.put(5, "five");
- context.setRecordContext(createRecordContext(startTime + incr * 6));
- store.put(6, "six");
- context.setRecordContext(createRecordContext(startTime + incr * 7));
- store.put(7, "seven");
- context.setRecordContext(createRecordContext(startTime + incr * 8));
- store.put(8, "eight");
- store.flush();
-
- } finally {
- store.close();
- }
-
+ context.setRecordContext(createRecordContext(startTime));
+ store.put(0, "zero");
+ context.setRecordContext(createRecordContext(startTime + incr));
+ store.put(1, "one");
+ context.setRecordContext(createRecordContext(startTime + incr * 2));
+ store.put(2, "two");
+ context.setRecordContext(createRecordContext(startTime + incr * 3));
+ store.put(3, "three");
+ context.setRecordContext(createRecordContext(startTime + incr * 4));
+ store.put(4, "four");
+ context.setRecordContext(createRecordContext(startTime + incr * 5));
+ store.put(5, "five");
+ context.setRecordContext(createRecordContext(startTime + incr * 6));
+ store.put(6, "six");
+ context.setRecordContext(createRecordContext(startTime + incr * 7));
+ store.put(7, "seven");
+ context.setRecordContext(createRecordContext(startTime + incr * 8));
+ store.put(8, "eight");
+ store.flush();
} finally {
- Utils.delete(baseDir);
+ store.close();
}
- File baseDir2 = Files.createTempDirectory("test").toFile();
- try {
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRestoreII") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- changeLog.add(new KeyValue<>(
- keySerializer.serialize(topic, key),
- valueSerializer.serialize(topic, value))
- );
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
-
- try {
- context.restore(windowName, changeLog);
-
- assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
- assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
- assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
- assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
- assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
- assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
- assertEquals(Utils.mkList("eight"), toList(store.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize)));
-
- // check segment directories
- store.flush();
- assertEquals(
- Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
- segmentDirs(baseDir)
- );
- } finally {
- store.close();
- }
+ // remove local store image
+ Utils.delete(baseDir);
+ WindowStore<Integer, String> store2 = createWindowStore(context, false, true);
+ assertEquals(Utils.mkList(), toList(store2.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize)));
+ try {
+ context.restore(windowName, changeLog);
+
+ assertEquals(Utils.mkList(), toList(store2.fetch(0, startTime - windowSize, startTime + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize)));
+ assertEquals(Utils.mkList(), toList(store2.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize)));
+ assertEquals(Utils.mkList("four"), toList(store2.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize)));
+ assertEquals(Utils.mkList("five"), toList(store2.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize)));
+ assertEquals(Utils.mkList("six"), toList(store2.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize)));
+ assertEquals(Utils.mkList("seven"), toList(store2.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize)));
+ assertEquals(Utils.mkList("eight"), toList(store2.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize)));
+
+ // check segment directories
+ store2.flush();
+ assertEquals(
+ Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
+ segmentDirs(baseDir)
+ );
} finally {
- Utils.delete(baseDir2);
+ store2.close();
}
}
@Test
public void testSegmentMaintenance() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
try {
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestSegmentMaintenance") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- // do nothing
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
- try {
-
- context.setTime(0L);
- context.setRecordContext(createRecordContext(0));
- store.put(0, "v");
- assertEquals(
- Utils.mkSet(segments.segmentName(0L)),
- segmentDirs(baseDir)
- );
-
- context.setRecordContext(createRecordContext(59999));
- store.put(0, "v");
- store.put(0, "v");
- assertEquals(
- Utils.mkSet(segments.segmentName(0L)),
- segmentDirs(baseDir)
- );
-
- context.setRecordContext(createRecordContext(60000));
- store.put(0, "v");
- assertEquals(
- Utils.mkSet(segments.segmentName(0L), segments.segmentName(1L)),
- segmentDirs(baseDir)
- );
-
- WindowStoreIterator iter;
- int fetchedCount;
-
- iter = store.fetch(0, 0L, 240000L);
- fetchedCount = 0;
- while (iter.hasNext()) {
- iter.next();
- fetchedCount++;
- }
- assertEquals(4, fetchedCount);
-
- assertEquals(
- Utils.mkSet(segments.segmentName(0L), segments.segmentName(1L)),
- segmentDirs(baseDir)
- );
-
- context.setRecordContext(createRecordContext(180000));
- store.put(0, "v");
+ context.setTime(0L);
+ context.setRecordContext(createRecordContext(0));
+ store.put(0, "v");
+ assertEquals(
+ Utils.mkSet(segments.segmentName(0L)),
+ segmentDirs(baseDir)
+ );
+
+ context.setRecordContext(createRecordContext(59999));
+ store.put(0, "v");
+ store.put(0, "v");
+ assertEquals(
+ Utils.mkSet(segments.segmentName(0L)),
+ segmentDirs(baseDir)
+ );
+
+ context.setRecordContext(createRecordContext(60000));
+ store.put(0, "v");
+ assertEquals(
+ Utils.mkSet(segments.segmentName(0L), segments.segmentName(1L)),
+ segmentDirs(baseDir)
+ );
+
+ WindowStoreIterator iter;
+ int fetchedCount;
+
+ iter = store.fetch(0, 0L, 240000L);
+ fetchedCount = 0;
+ while (iter.hasNext()) {
+ iter.next();
+ fetchedCount++;
+ }
+ assertEquals(4, fetchedCount);
- iter = store.fetch(0, 0L, 240000L);
- fetchedCount = 0;
- while (iter.hasNext()) {
- iter.next();
- fetchedCount++;
- }
- assertEquals(2, fetchedCount);
+ assertEquals(
+ Utils.mkSet(segments.segmentName(0L), segments.segmentName(1L)),
+ segmentDirs(baseDir)
+ );
- assertEquals(
- Utils.mkSet(segments.segmentName(1L), segments.segmentName(3L)),
- segmentDirs(baseDir)
- );
+ context.setRecordContext(createRecordContext(180000));
+ store.put(0, "v");
- context.setRecordContext(createRecordContext(300000));
- store.put(0, "v");
+ iter = store.fetch(0, 0L, 240000L);
+ fetchedCount = 0;
+ while (iter.hasNext()) {
+ iter.next();
+ fetchedCount++;
+ }
+ assertEquals(2, fetchedCount);
- iter = store.fetch(0, 240000L, 1000000L);
- fetchedCount = 0;
- while (iter.hasNext()) {
- iter.next();
- fetchedCount++;
- }
- assertEquals(1, fetchedCount);
+ assertEquals(
+ Utils.mkSet(segments.segmentName(1L), segments.segmentName(3L)),
+ segmentDirs(baseDir)
+ );
- assertEquals(
- Utils.mkSet(segments.segmentName(3L), segments.segmentName(5L)),
- segmentDirs(baseDir)
- );
+ context.setRecordContext(createRecordContext(300000));
+ store.put(0, "v");
- } finally {
- store.close();
+ iter = store.fetch(0, 240000L, 1000000L);
+ fetchedCount = 0;
+ while (iter.hasNext()) {
+ iter.next();
+ fetchedCount++;
}
+ assertEquals(1, fetchedCount);
+
+ assertEquals(
+ Utils.mkSet(segments.segmentName(3L), segments.segmentName(5L)),
+ segmentDirs(baseDir)
+ );
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
@Test
public void testInitialLoading() throws IOException {
- File baseDir = Files.createTempDirectory("test").toFile();
- try {
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestInitialLoading") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- // do nothing
- }
- };
+ File storeDir = new File(baseDir, windowName);
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
+ WindowStore<Integer, String> store = createWindowStore(context, false, true);
- File storeDir = new File(baseDir, windowName);
-
- WindowStore<Integer, String> store = createWindowStore(context, false, true);
-
-
- try {
- new File(storeDir, segments.segmentName(0L)).mkdir();
- new File(storeDir, segments.segmentName(1L)).mkdir();
- new File(storeDir, segments.segmentName(2L)).mkdir();
- new File(storeDir, segments.segmentName(3L)).mkdir();
- new File(storeDir, segments.segmentName(4L)).mkdir();
- new File(storeDir, segments.segmentName(5L)).mkdir();
- new File(storeDir, segments.segmentName(6L)).mkdir();
- } finally {
- store.close();
- }
+ try {
+ new File(storeDir, segments.segmentName(0L)).mkdir();
+ new File(storeDir, segments.segmentName(1L)).mkdir();
+ new File(storeDir, segments.segmentName(2L)).mkdir();
+ new File(storeDir, segments.segmentName(3L)).mkdir();
+ new File(storeDir, segments.segmentName(4L)).mkdir();
+ new File(storeDir, segments.segmentName(5L)).mkdir();
+ new File(storeDir, segments.segmentName(6L)).mkdir();
+ } finally {
+ store.close();
+ }
- store = createWindowStore(context, false, true);
+ store = createWindowStore(context, false, true);
- try {
- assertEquals(
- Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
- segmentDirs(baseDir)
- );
+ try {
+ assertEquals(
+ Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
+ segmentDirs(baseDir)
+ );
- try (WindowStoreIterator iter = store.fetch(0, 0L, 1000000L)) {
- while (iter.hasNext()) {
- iter.next();
- }
+ try (WindowStoreIterator iter = store.fetch(0, 0L, 1000000L)) {
+ while (iter.hasNext()) {
+ iter.next();
}
-
- assertEquals(
- Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
- segmentDirs(baseDir)
- );
-
- } finally {
- store.close();
}
+ assertEquals(
+ Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)),
+ segmentDirs(baseDir)
+ );
+
} finally {
- Utils.delete(baseDir);
+ store.close();
}
}
@Test
public void shouldCloseOpenIteratorsWhenStoreIsClosedAndThrowInvalidStateStoreExceptionOnHasNextAndNext() throws Exception {
- final File baseDir = TestUtils.tempDirectory();
- Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer());
- RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-ShouldOnlyIterateOpenSegments") {
- @Override
- public <K1, V1> void send(final String topic,
- K1 key,
- V1 value,
- Integer partition,
- Long timestamp,
- Serializer<K1> keySerializer,
- Serializer<V1> valueSerializer) {
- }
- };
-
- MockProcessorContext context = new MockProcessorContext(
- baseDir,
- byteArraySerde, byteArraySerde,
- recordCollector, cache);
-
final WindowStore<Integer, String> windowStore = createWindowStore(context, false, true);
context.setRecordContext(createRecordContext(0));
windowStore.put(1, "one", 1L);
@@ -969,6 +640,34 @@ public class RocksDBWindowStoreTest {
}
}
+ private void putFirstBatch(final WindowStore<Integer, String> store, final long startTime, final MockProcessorContext context) {
+ context.setRecordContext(createRecordContext(startTime));
+ store.put(0, "zero");
+ context.setRecordContext(createRecordContext(startTime + 1L));
+ store.put(1, "one");
+ context.setRecordContext(createRecordContext(startTime + 2L));
+ store.put(2, "two");
+ context.setRecordContext(createRecordContext(startTime + 4L));
+ store.put(4, "four");
+ context.setRecordContext(createRecordContext(startTime + 5L));
+ store.put(5, "five");
+ }
+
+ private void putSecondBatch(final WindowStore<Integer, String> store, final long startTime, MockProcessorContext context) {
+ context.setRecordContext(createRecordContext(startTime + 3L));
+ store.put(2, "two+1");
+ context.setRecordContext(createRecordContext(startTime + 4L));
+ store.put(2, "two+2");
+ context.setRecordContext(createRecordContext(startTime + 5L));
+ store.put(2, "two+3");
+ context.setRecordContext(createRecordContext(startTime + 6L));
+ store.put(2, "two+4");
+ context.setRecordContext(createRecordContext(startTime + 7L));
+ store.put(2, "two+5");
+ context.setRecordContext(createRecordContext(startTime + 8L));
+ store.put(2, "two+6");
+ }
+
private <E> List<E> toList(WindowStoreIterator<E> iterator) {
ArrayList<E> list = new ArrayList<>();
while (iterator.hasNext()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
index ae6fb5a..346d3de 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/SegmentIteratorTest.java
@@ -79,9 +79,7 @@ public class SegmentIteratorTest {
@Test
public void shouldIterateOverAllSegments() throws Exception {
- iterator = new SegmentIterator(
- Arrays.asList(segmentOne,
- segmentTwo).iterator(),
+ iterator = new SegmentIterator(Arrays.asList(segmentOne, segmentTwo).iterator(),
hasNextCondition,
Bytes.wrap("a".getBytes()),
Bytes.wrap("z".getBytes()));
@@ -107,9 +105,7 @@ public class SegmentIteratorTest {
@Test
public void shouldOnlyIterateOverSegmentsInRange() throws Exception {
- iterator = new SegmentIterator(
- Arrays.asList(segmentOne,
- segmentTwo).iterator(),
+ iterator = new SegmentIterator(Arrays.asList(segmentOne, segmentTwo).iterator(),
hasNextCondition,
Bytes.wrap("a".getBytes()),
Bytes.wrap("b".getBytes()));
@@ -127,9 +123,7 @@ public class SegmentIteratorTest {
@Test(expected = NoSuchElementException.class)
public void shouldThrowNoSuchElementOnPeekNextKeyIfNoNext() throws Exception {
- iterator = new SegmentIterator(
- Arrays.asList(segmentOne,
- segmentTwo).iterator(),
+ iterator = new SegmentIterator(Arrays.asList(segmentOne, segmentTwo).iterator(),
hasNextCondition,
Bytes.wrap("f".getBytes()),
Bytes.wrap("h".getBytes()));
@@ -139,9 +133,7 @@ public class SegmentIteratorTest {
@Test(expected = NoSuchElementException.class)
public void shouldThrowNoSuchElementOnNextIfNoNext() throws Exception {
- iterator = new SegmentIterator(
- Arrays.asList(segmentOne,
- segmentTwo).iterator(),
+ iterator = new SegmentIterator(Arrays.asList(segmentOne, segmentTwo).iterator(),
hasNextCondition,
Bytes.wrap("f".getBytes()),
Bytes.wrap("h".getBytes()));
[4/5] kafka git commit: KAFKA-3452 Follow-up: Optimize ByteStore
Scenarios
Posted by gu...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
index a8ddc73..6c5efcb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
@@ -17,8 +17,8 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.serialization.Serde;
+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.processor.ProcessorContext;
@@ -28,116 +28,78 @@ import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.StateSerdes;
-class RocksDBSessionStore<K, AGG> implements SessionStore<K, AGG> {
+class RocksDBSessionStore<K, AGG> extends WrappedStateStore.AbstractStateStore implements SessionStore<K, AGG> {
private final Serde<K> keySerde;
private final Serde<AGG> aggSerde;
- private final SegmentedBytesStore bytesStore;
- private StateSerdes<K, AGG> serdes;
+ protected final SegmentedBytesStore bytesStore;
+ protected StateSerdes<K, AGG> serdes;
- RocksDBSessionStore(final SegmentedBytesStore bytesStore,
- final Serde<K> keySerde,
- final Serde<AGG> aggSerde) {
- this.keySerde = keySerde;
- this.bytesStore = bytesStore;
- this.aggSerde = aggSerde;
- }
-
+ // this is optimizing the case when this store is already a bytes store, in which we can avoid Bytes.wrap() costs
+ private static class RocksDBSessionBytesStore extends RocksDBSessionStore<Bytes, byte[]> {
+ RocksDBSessionBytesStore(final SegmentedBytesStore inner) {
+ super(inner, Serdes.Bytes(), Serdes.ByteArray());
+ }
- @SuppressWarnings("unchecked")
- @Override
- public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), earliestSessionEndTime, latestSessionStartTime);
- return new SessionStoreIterator(bytesIterator, serdes);
- }
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes key, final long earliestSessionEndTime, final long latestSessionStartTime) {
+ final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(key, earliestSessionEndTime, latestSessionStartTime);
+ return WrappedSessionStoreIterator.bytesIterator(bytesIterator, serdes);
+ }
+ @Override
+ public void remove(final Windowed<Bytes> key) {
+ bytesStore.remove(SessionKeySerde.bytesToBinary(key));
+ }
- @Override
- public void remove(final Windowed<K> key) {
- bytesStore.remove(SessionKeySerde.toBinary(key, serdes.keySerializer()));
+ @Override
+ public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
+ bytesStore.put(SessionKeySerde.bytesToBinary(sessionKey), aggregate);
+ }
}
- @Override
- public void put(final Windowed<K> sessionKey, final AGG aggregate) {
- bytesStore.put(SessionKeySerde.toBinary(sessionKey, serdes.keySerializer()), aggSerde.serializer().serialize(bytesStore.name(), aggregate));
+ static RocksDBSessionStore<Bytes, byte[]> bytesStore(final SegmentedBytesStore inner) {
+ return new RocksDBSessionBytesStore(inner);
}
- @Override
- public String name() {
- return bytesStore.name();
+ RocksDBSessionStore(final SegmentedBytesStore bytesStore,
+ final Serde<K> keySerde,
+ final Serde<AGG> aggSerde) {
+ super(bytesStore);
+ this.keySerde = keySerde;
+ this.bytesStore = bytesStore;
+ this.aggSerde = aggSerde;
}
@Override
@SuppressWarnings("unchecked")
public void init(final ProcessorContext context, final StateStore root) {
this.serdes = new StateSerdes<>(bytesStore.name(),
- keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
- aggSerde == null ? (Serde<AGG>) context.valueSerde() : aggSerde);
+ keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
+ aggSerde == null ? (Serde<AGG>) context.valueSerde() : aggSerde);
bytesStore.init(context, root);
}
@Override
- public void flush() {
- bytesStore.flush();
- }
-
- @Override
- public void close() {
- bytesStore.close();
+ public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) {
+ final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), earliestSessionEndTime, latestSessionStartTime);
+ return new WrappedSessionStoreIterator<>(bytesIterator, serdes);
}
@Override
- public boolean persistent() {
- return true;
+ public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) {
+ return findSessions(key, 0, Long.MAX_VALUE);
}
@Override
- public boolean isOpen() {
- return bytesStore.isOpen();
+ public void remove(final Windowed<K> key) {
+ bytesStore.remove(SessionKeySerde.toBinary(key, serdes.keySerializer()));
}
@Override
- public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) {
- return findSessions(key, 0, Long.MAX_VALUE);
- }
-
- private static class SessionStoreIterator<K, AGG> implements KeyValueIterator<Windowed<K>, AGG> {
-
- private final KeyValueIterator<Bytes, byte[]> bytesIterator;
- private final StateSerdes<K, AGG> serdes;
-
- SessionStoreIterator(final KeyValueIterator<Bytes, byte[]> bytesIterator, final StateSerdes<K, AGG> serdes) {
- this.bytesIterator = bytesIterator;
- this.serdes = serdes;
- }
-
- @Override
- public void close() {
- bytesIterator.close();
- }
-
- @Override
- public Windowed<K> peekNextKey() {
- final Bytes bytes = bytesIterator.peekNextKey();
- return SessionKeySerde.from(bytes.get(), serdes.keyDeserializer());
- }
-
- @Override
- public boolean hasNext() {
- return bytesIterator.hasNext();
- }
-
- @Override
- public KeyValue<Windowed<K>, AGG> next() {
- final KeyValue<Bytes, byte[]> next = bytesIterator.next();
- return KeyValue.pair(SessionKeySerde.from(next.key.get(), serdes.keyDeserializer()), serdes.valueFrom(next.value));
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException("remove not supported by SessionStoreIterator");
- }
+ public void put(final Windowed<K> sessionKey, final AGG aggregate) {
+ bytesStore.put(SessionKeySerde.toBinary(sessionKey, serdes.keySerializer()), aggSerde.serializer().serialize(bytesStore.name(), aggregate));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplier.java
index 10ebf65..9eb9447 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplier.java
@@ -18,7 +18,6 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.state.SessionStore;
@@ -33,11 +32,10 @@ import java.util.Map;
*
* @see org.apache.kafka.streams.state.Stores#create(String)
*/
-
public class RocksDBSessionStoreSupplier<K, V> extends AbstractStoreSupplier<K, V, SessionStore> implements WindowStoreSupplier<SessionStore> {
+ private static final String METRIC_SCOPE = "rocksdb-session";
private static final int NUM_SEGMENTS = 3;
- public static final String METRIC_SCOPE = "rocksdb-session-store";
private final long retentionPeriod;
private final boolean cached;
@@ -56,15 +54,14 @@ public class RocksDBSessionStoreSupplier<K, V> extends AbstractStoreSupplier<K,
final RocksDBSegmentedBytesStore segmented = new RocksDBSegmentedBytesStore(name,
retentionPeriod,
NUM_SEGMENTS,
- keySchema
- );
+ keySchema);
if (cached && logged) {
final ChangeLoggingSegmentedBytesStore logged = new ChangeLoggingSegmentedBytesStore(segmented);
final MeteredSegmentedBytesStore metered = new MeteredSegmentedBytesStore(logged,
METRIC_SCOPE, time);
final RocksDBSessionStore<Bytes, byte[]> sessionStore
- = new RocksDBSessionStore<>(metered, Serdes.Bytes(), Serdes.ByteArray());
+ = RocksDBSessionStore.bytesStore(metered);
return new CachingSessionStore<>(sessionStore, keySerde, valueSerde);
}
@@ -73,7 +70,7 @@ public class RocksDBSessionStoreSupplier<K, V> extends AbstractStoreSupplier<K,
final MeteredSegmentedBytesStore metered = new MeteredSegmentedBytesStore(segmented,
METRIC_SCOPE, time);
final RocksDBSessionStore<Bytes, byte[]> sessionStore
- = new RocksDBSessionStore<>(metered, Serdes.Bytes(), Serdes.ByteArray());
+ = RocksDBSessionStore.bytesStore(metered);
return new CachingSessionStore<>(sessionStore, keySerde, valueSerde);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index 280d5c4..108c772 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -152,12 +152,12 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
// value getter should always read directly from rocksDB
// since it is only for values that are already flushed
context.register(root, false, new StateRestoreCallback() {
-
@Override
public void restore(byte[] key, byte[] value) {
putInternal(key, value);
}
});
+
open = true;
}
@@ -283,7 +283,7 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
public synchronized KeyValueIterator<K, V> range(K from, K to) {
validateStoreOpen();
// query rocksdb
- final RocksDBRangeIterator rocksDBRangeIterator = new RocksDBRangeIterator(db.newIterator(), serdes, from, to);
+ final RocksDBRangeIterator rocksDBRangeIterator = new RocksDBRangeIterator(name, db.newIterator(), serdes, from, to);
openIterators.add(rocksDBRangeIterator);
return rocksDBRangeIterator;
}
@@ -294,7 +294,7 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
// query rocksdb
RocksIterator innerIter = db.newIterator();
innerIter.seekToFirst();
- final RocksDbIterator rocksDbIterator = new RocksDbIterator(innerIter, serdes);
+ final RocksDbIterator rocksDbIterator = new RocksDbIterator(name, innerIter, serdes);
openIterators.add(rocksDbIterator);
return rocksDbIterator;
}
@@ -312,6 +312,7 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
*/
@Override
public long approximateNumEntries() {
+ validateStoreOpen();
long value;
try {
value = this.db.getLongProperty("rocksdb.estimate-num-keys");
@@ -375,15 +376,17 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
openIterators.clear();
}
-
- class RocksDbIterator implements KeyValueIterator<K, V> {
+ private class RocksDbIterator implements KeyValueIterator<K, V> {
+ private final String storeName;
private final RocksIterator iter;
private final StateSerdes<K, V> serdes;
- private boolean open = true;
- RocksDbIterator(RocksIterator iter, StateSerdes<K, V> serdes) {
+ private volatile boolean open = true;
+
+ RocksDbIterator(String storeName, RocksIterator iter, StateSerdes<K, V> serdes) {
this.iter = iter;
this.serdes = serdes;
+ this.storeName = storeName;
}
byte[] peekRawKey() {
@@ -397,8 +400,9 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
@Override
public synchronized boolean hasNext() {
if (!open) {
- throw new InvalidStateStoreException("store %s has closed");
+ throw new InvalidStateStoreException(String.format("RocksDB store %s has closed", storeName));
}
+
return iter.isValid();
}
@@ -415,19 +419,16 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
return entry;
}
- /**
- * @throws UnsupportedOperationException
- */
@Override
public void remove() {
- throw new UnsupportedOperationException("RocksDB iterator does not support remove");
+ throw new UnsupportedOperationException("RocksDB iterator does not support remove()");
}
@Override
public synchronized void close() {
- open = false;
openIterators.remove(this);
iter.close();
+ open = false;
}
@Override
@@ -436,9 +437,7 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
throw new NoSuchElementException();
}
return serdes.keyFrom(iter.key());
-
}
-
}
private class RocksDBRangeIterator extends RocksDbIterator {
@@ -448,8 +447,8 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
private final Comparator<byte[]> comparator = Bytes.BYTES_LEXICO_COMPARATOR;
private byte[] rawToKey;
- RocksDBRangeIterator(RocksIterator iter, StateSerdes<K, V> serdes, K from, K to) {
- super(iter, serdes);
+ RocksDBRangeIterator(String storeName, RocksIterator iter, StateSerdes<K, V> serdes, K from, K to) {
+ super(storeName, iter, serdes);
iter.seek(serdes.rawKey(from));
this.rawToKey = serdes.rawKey(to);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
index 80c4796..8c7a557 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
@@ -22,7 +22,6 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.serialization.Serde;
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.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -30,38 +29,50 @@ import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
-import java.util.NoSuchElementException;
+class RocksDBWindowStore<K, V> extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V> {
-class RocksDBWindowStore<K, V> implements WindowStore<K, V> {
-
- private final SegmentedBytesStore bytesStore;
- private final boolean retainDuplicates;
private final Serde<K> keySerde;
private final Serde<V> valueSerde;
+ protected final SegmentedBytesStore bytesStore;
+ protected final boolean retainDuplicates;
+
private ProcessorContext context;
- private int seqnum = 0;
- private StateSerdes<K, V> serdes;
+ protected StateSerdes<K, V> serdes;
+ protected int seqnum = 0;
+ // this is optimizing the case when this store is already a bytes store, in which we can avoid Bytes.wrap() costs
+ private static class RocksDBWindowBytesStore extends RocksDBWindowStore<Bytes, byte[]> {
+ RocksDBWindowBytesStore(final SegmentedBytesStore inner, final boolean retainDuplicates) {
+ super(inner, Serdes.Bytes(), Serdes.ByteArray(), retainDuplicates);
+ }
- static RocksDBWindowStore<Bytes, byte[]> bytesStore(final SegmentedBytesStore inner, final boolean retainDuplicates) {
- return new RocksDBWindowStore<>(inner, Serdes.Bytes(), Serdes.ByteArray(), retainDuplicates);
+ @Override
+ public void put(Bytes key, byte[] value, long timestamp) {
+ maybeUpdateSeqnumForDups();
+
+ bytesStore.put(WindowStoreUtils.toBinaryKey(key.get(), timestamp, seqnum), value);
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> fetch(Bytes key, long timeFrom, long timeTo) {
+ final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(key, timeFrom, timeTo);
+ return WrappedWindowStoreIterator.bytesIterator(bytesIterator, serdes);
+ }
}
+ static RocksDBWindowStore<Bytes, byte[]> bytesStore(final SegmentedBytesStore inner, final boolean retainDuplicates) {
+ return new RocksDBWindowBytesStore(inner, retainDuplicates);
+ }
RocksDBWindowStore(final SegmentedBytesStore bytesStore,
final Serde<K> keySerde,
final Serde<V> valueSerde,
final boolean retainDuplicates) {
+ super(bytesStore);
this.keySerde = keySerde;
this.valueSerde = valueSerde;
- this.retainDuplicates = retainDuplicates;
this.bytesStore = bytesStore;
- }
-
-
- @Override
- public String name() {
- return bytesStore.name();
+ this.retainDuplicates = retainDuplicates;
}
@Override
@@ -77,91 +88,26 @@ class RocksDBWindowStore<K, V> implements WindowStore<K, V> {
}
@Override
- public boolean persistent() {
- return true;
- }
-
- @Override
- public boolean isOpen() {
- return bytesStore.isOpen();
- }
-
- @Override
- public void flush() {
- bytesStore.flush();
- }
-
- @Override
- public void close() {
- bytesStore.close();
- }
-
- @Override
public void put(K key, V value) {
put(key, value, context.timestamp());
}
@Override
public void put(K key, V value, long timestamp) {
- if (retainDuplicates) {
- seqnum = (seqnum + 1) & 0x7FFFFFFF;
- }
- bytesStore.put(Bytes.wrap(WindowStoreUtils.toBinaryKey(key, timestamp, seqnum, serdes)), serdes.rawValue(value));
- }
+ maybeUpdateSeqnumForDups();
+ bytesStore.put(WindowStoreUtils.toBinaryKey(key, timestamp, seqnum, serdes), serdes.rawValue(value));
+ }
- @SuppressWarnings("unchecked")
@Override
public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) {
final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo);
- return new TheWindowStoreIterator<>(bytesIterator, serdes);
+ return new WrappedWindowStoreIterator<>(bytesIterator, serdes);
}
- private static class TheWindowStoreIterator<V> implements WindowStoreIterator<V> {
- private final KeyValueIterator<Bytes, byte[]> actual;
- private final StateSerdes<?, V> serdes;
-
- TheWindowStoreIterator(final KeyValueIterator<Bytes, byte[]> actual, final StateSerdes<?, V> serdes) {
- this.actual = actual;
- this.serdes = serdes;
- }
-
- @Override
- public boolean hasNext() {
- return actual.hasNext();
- }
-
- /**
- * @throws NoSuchElementException if no next element exists
- */
- @Override
- public KeyValue<Long, V> next() {
- if (!actual.hasNext()) {
- throw new NoSuchElementException();
- }
- final KeyValue<Bytes, byte[]> next = actual.next();
- final long timestamp = WindowStoreUtils.timestampFromBinaryKey(next.key.get());
- final V value = serdes.valueFrom(next.value);
- return KeyValue.pair(timestamp, value);
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() {
- actual.close();
- }
-
- @Override
- public Long peekNextKey() {
- if (!actual.hasNext()) {
- throw new NoSuchElementException();
- }
- return WindowStoreUtils.timestampFromBinaryKey(actual.peekNextKey().get());
+ void maybeUpdateSeqnumForDups() {
+ if (retainDuplicates) {
+ seqnum = (seqnum + 1) & 0x7FFFFFFF;
}
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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 abaaffd..9782226 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
@@ -65,7 +65,7 @@ public class RocksDBWindowStoreSupplier<K, V> extends AbstractStoreSupplier<K, V
name,
retentionPeriod,
numSegments,
- new WindowStoreKeySchema()
+ new WindowKeySchema()
)));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/Segment.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/Segment.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/Segment.java
index 5471262..5bbdaab 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/Segment.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/Segment.java
@@ -20,7 +20,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.processor.ProcessorContext;
-// use the Bytes wrapper for underlying rocksDB keys since they are used for hashing data structures
+// Use the Bytes wrapper for underlying rocksDB keys since they are used for hashing data structures
class Segment extends RocksDBStore<Bytes, byte[]> {
public final long id;
@@ -29,14 +29,16 @@ class Segment extends RocksDBStore<Bytes, byte[]> {
this.id = id;
}
- public void destroy() {
+ void destroy() {
Utils.delete(dbDir);
}
@Override
public void openDB(final ProcessorContext context) {
super.openDB(context);
+
+ // skip the registering step
+
open = true;
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
index e6715b2..c449fdb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
@@ -30,12 +30,13 @@ import java.util.NoSuchElementException;
*/
class SegmentIterator implements KeyValueIterator<Bytes, byte[]> {
- private final Iterator<Segment> segments;
- private final HasNextCondition hasNextCondition;
private final Bytes from;
private final Bytes to;
- private KeyValueIterator<Bytes, byte[]> currentIterator;
+ private final Iterator<Segment> segments;
+ private final HasNextCondition hasNextCondition;
+
private KeyValueStore<Bytes, byte[]> currentSegment;
+ private KeyValueIterator<Bytes, byte[]> currentIterator;
SegmentIterator(final Iterator<Segment> segments,
final HasNextCondition hasNextCondition,
@@ -62,7 +63,7 @@ class SegmentIterator implements KeyValueIterator<Bytes, byte[]> {
return currentIterator.peekNextKey();
}
- @SuppressWarnings("unchecked")
+ @Override
public boolean hasNext() {
boolean hasNext = false;
while ((currentIterator == null || !(hasNext = hasNextCondition.hasNext(currentIterator)) || !currentSegment.isOpen())
@@ -86,7 +87,6 @@ class SegmentIterator implements KeyValueIterator<Bytes, byte[]> {
}
public void remove() {
- throw new UnsupportedOperationException("remove not supported");
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentedBytesStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentedBytesStore.java
index ab1099e..ab08fe6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentedBytesStore.java
@@ -65,13 +65,12 @@ public interface SegmentedBytesStore extends StateStore {
*/
byte[] get(Bytes key);
-
interface KeySchema {
/**
* Given a record-key and a time, construct a Segmented key that represents
* the upper range of keys to search when performing range queries.
* @see SessionKeySchema#upperRange
- * @see WindowStoreKeySchema#upperRange
+ * @see WindowKeySchema#upperRange
* @param key
* @param to
* @return The key that represents the upper range to search for in the store
@@ -82,7 +81,7 @@ public interface SegmentedBytesStore extends StateStore {
* Given a record-key and a time, construct a Segmented key that represents
* the lower range of keys to search when performing range queries.
* @see SessionKeySchema#lowerRange
- * @see WindowStoreKeySchema#lowerRange
+ * @see WindowKeySchema#lowerRange
* @param key
* @param from
* @return The key that represents the lower range to search for in the store
@@ -93,7 +92,7 @@ public interface SegmentedBytesStore extends StateStore {
* Extract the timestamp of the segment from the key. The key is a composite of
* the record-key, any timestamps, plus any additional information.
* @see SessionKeySchema#lowerRange
- * @see WindowStoreKeySchema#lowerRange
+ * @see WindowKeySchema#lowerRange
* @param key
* @return
*/
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/Segments.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/Segments.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/Segments.java
index 13cdf6c..bc2f4fe 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/Segments.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/Segments.java
@@ -35,11 +35,13 @@ import java.util.concurrent.ConcurrentHashMap;
*/
class Segments {
static final long MIN_SEGMENT_INTERVAL = 60 * 1000L;
+
private final ConcurrentHashMap<Long, Segment> segments = new ConcurrentHashMap<>();
private final String name;
private final int numSegments;
private final long segmentInterval;
private final SimpleDateFormat formatter;
+
private long currentSegmentId = -1L;
Segments(final String name, final long retentionPeriod, final int numSegments) {
@@ -82,7 +84,6 @@ class Segments {
}
}
-
void openExisting(final ProcessorContext context) {
try {
File dir = new File(context.stateDir(), name);
@@ -103,7 +104,7 @@ class Segments {
}
} else {
if (!dir.mkdir()) {
- throw new ProcessorStateException(String.format("dir %s doesn't exist and cannot be created for store %s", dir, name));
+ throw new ProcessorStateException(String.format("dir %s doesn't exist and cannot be created for segments %s", dir, name));
}
}
} catch (Exception ex) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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
index d76e8a4..e36e567 100644
--- 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
@@ -65,6 +65,6 @@ class SerializedKeyValueIterator<K, V> implements KeyValueIterator<K, V> {
@Override
public void remove() {
- throw new UnsupportedOperationException("remove not supported by SerializedKeyValueIterator");
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
index f8f0e08..d47b636 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
@@ -38,10 +38,11 @@ import java.util.NoSuchElementException;
*/
public class ThreadCache {
private static final Logger log = LoggerFactory.getLogger(ThreadCache.class);
+
private final String name;
private final long maxCacheSizeBytes;
- private final Map<String, NamedCache> caches = new HashMap<>();
private final StreamsMetrics metrics;
+ private final Map<String, NamedCache> caches = new HashMap<>();
// internal stats
private long numPuts = 0;
@@ -49,8 +50,6 @@ public class ThreadCache {
private long numEvicts = 0;
private long numFlushes = 0;
-
-
public interface DirtyEntryFlushListener {
void apply(final List<DirtyEntry> dirty);
}
@@ -101,7 +100,7 @@ public class ThreadCache {
name, puts(), gets(), evicts(), flushes());
}
- public LRUCacheEntry get(final String namespace, byte[] key) {
+ public LRUCacheEntry get(final String namespace, Bytes key) {
numGets++;
if (key == null) {
@@ -112,20 +111,21 @@ public class ThreadCache {
if (cache == null) {
return null;
}
- return cache.get(Bytes.wrap(key));
+ return cache.get(key);
}
- public void put(final String namespace, byte[] key, LRUCacheEntry value) {
+ public void put(final String namespace, Bytes key, LRUCacheEntry value) {
numPuts++;
+
final NamedCache cache = getOrCreateCache(namespace);
- cache.put(Bytes.wrap(key), value);
+ cache.put(key, value);
maybeEvict(namespace);
}
- public LRUCacheEntry putIfAbsent(final String namespace, byte[] key, LRUCacheEntry value) {
+ public LRUCacheEntry putIfAbsent(final String namespace, Bytes key, LRUCacheEntry value) {
final NamedCache cache = getOrCreateCache(namespace);
- final LRUCacheEntry result = cache.putIfAbsent(Bytes.wrap(key), value);
+ final LRUCacheEntry result = cache.putIfAbsent(key, value);
maybeEvict(namespace);
if (result == null) {
@@ -134,27 +134,27 @@ public class ThreadCache {
return result;
}
- public void putAll(final String namespace, final List<KeyValue<byte[], LRUCacheEntry>> entries) {
- for (KeyValue<byte[], LRUCacheEntry> entry : entries) {
+ public void putAll(final String namespace, final List<KeyValue<Bytes, LRUCacheEntry>> entries) {
+ for (KeyValue<Bytes, LRUCacheEntry> entry : entries) {
put(namespace, entry.key, entry.value);
}
}
- public LRUCacheEntry delete(final String namespace, final byte[] key) {
+ public LRUCacheEntry delete(final String namespace, final Bytes key) {
final NamedCache cache = getCache(namespace);
if (cache == null) {
return null;
}
- return cache.delete(Bytes.wrap(key));
+ return cache.delete(key);
}
- public MemoryLRUCacheBytesIterator range(final String namespace, final byte[] from, final byte[] to) {
+ public MemoryLRUCacheBytesIterator range(final String namespace, final Bytes from, final Bytes to) {
final NamedCache cache = getCache(namespace);
if (cache == null) {
return new MemoryLRUCacheBytesIterator(Collections.<Bytes>emptyIterator(), new NamedCache(namespace, this.metrics));
}
- return new MemoryLRUCacheBytesIterator(cache.keyRange(cacheKey(from), cacheKey(to)), cache);
+ return new MemoryLRUCacheBytesIterator(cache.keyRange(from, to), cache);
}
public MemoryLRUCacheBytesIterator all(final String namespace) {
@@ -229,11 +229,6 @@ public class ThreadCache {
return cache;
}
- private Bytes cacheKey(final byte[] keyBytes) {
- return Bytes.wrap(keyBytes);
- }
-
-
static class MemoryLRUCacheBytesIterator implements PeekingKeyValueIterator<Bytes, LRUCacheEntry> {
private final Iterator<Bytes> keys;
private final NamedCache cache;
@@ -303,12 +298,12 @@ public class ThreadCache {
}
}
- public static class DirtyEntry {
+ static class DirtyEntry {
private final Bytes key;
private final byte[] newValue;
private final RecordContext recordContext;
- public DirtyEntry(final Bytes key, final byte[] newValue, final RecordContext recordContext) {
+ DirtyEntry(final Bytes key, final byte[] newValue, final RecordContext recordContext) {
this.key = key;
this.newValue = newValue;
this.recordContext = recordContext;
@@ -326,5 +321,4 @@ public class ThreadCache {
return recordContext;
}
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowKeySchema.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowKeySchema.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowKeySchema.java
new file mode 100644
index 0000000..ea607c0
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowKeySchema.java
@@ -0,0 +1,59 @@
+/**
+ * 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.state.KeyValueIterator;
+import org.apache.kafka.streams.state.StateSerdes;
+
+import java.util.List;
+
+class WindowKeySchema implements RocksDBSegmentedBytesStore.KeySchema {
+ private static final HasNextCondition ITERATOR_HAS_NEXT = new HasNextCondition() {
+ @Override
+ public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) {
+ return iterator.hasNext();
+ }
+ };
+ private final StateSerdes<Bytes, byte[]> serdes = new StateSerdes<>("window-store-key-schema", Serdes.Bytes(), Serdes.ByteArray());
+
+ @Override
+ public Bytes upperRange(final Bytes key, final long to) {
+ return WindowStoreUtils.toBinaryKey(key, to, Integer.MAX_VALUE, serdes);
+ }
+
+ @Override
+ public Bytes lowerRange(final Bytes key, final long from) {
+ return WindowStoreUtils.toBinaryKey(key, Math.max(0, from), 0, serdes);
+ }
+
+ @Override
+ public long segmentTimestamp(final Bytes key) {
+ return WindowStoreUtils.timestampFromBinaryKey(key.get());
+ }
+
+ @Override
+ public HasNextCondition hasNextCondition(final Bytes binaryKey, final long from, final long to) {
+ return ITERATOR_HAS_NEXT;
+ }
+
+ @Override
+ public List<Segment> segmentsToSearch(final Segments segments, final long from, final long to) {
+ return segments.segments(from, to);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreKeySchema.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreKeySchema.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreKeySchema.java
deleted file mode 100644
index 093161e..0000000
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreKeySchema.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/**
- * 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.state.KeyValueIterator;
-import org.apache.kafka.streams.state.StateSerdes;
-
-import java.util.List;
-
-class WindowStoreKeySchema implements RocksDBSegmentedBytesStore.KeySchema {
- private static final HasNextCondition ITERATOR_HAS_NEXT = new HasNextCondition() {
- @Override
- public boolean hasNext(final KeyValueIterator<Bytes, ?> iterator) {
- return iterator.hasNext();
- }
- };
- private final StateSerdes<Bytes, byte[]> serdes = new StateSerdes<>("window-store-key-schema", Serdes.Bytes(), Serdes.ByteArray());
-
- @Override
- public Bytes upperRange(final Bytes key, final long to) {
- return Bytes.wrap(WindowStoreUtils.toBinaryKey(key, to, Integer.MAX_VALUE, serdes));
- }
-
- @Override
- public Bytes lowerRange(final Bytes key, final long from) {
- return Bytes.wrap(WindowStoreUtils.toBinaryKey(key, Math.max(0, from), 0, serdes));
- }
-
- @Override
- public long segmentTimestamp(final Bytes key) {
- return WindowStoreUtils.timestampFromBinaryKey(key.get());
- }
-
- @Override
- public HasNextCondition hasNextCondition(final Bytes binaryKey, final long from, final long to) {
- return ITERATOR_HAS_NEXT;
- }
-
- @Override
- public List<Segment> segmentsToSearch(final Segments segments, final long from, final long to) {
- return segments.segments(from, to);
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreSupplier.java
index 9c93dcc..8d0d673 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreSupplier.java
@@ -19,6 +19,13 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreSupplier;
+/**
+ * A windowed state store supplier that extends the {@link StateStoreSupplier} interface.
+ *
+ * @param <T> State store type
+ */
public interface WindowStoreSupplier<T extends StateStore> extends StateStoreSupplier<T> {
+
+ // window retention period in milli-second
long retentionPeriod();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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 074cf8a..d97b053 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
@@ -32,26 +32,25 @@ public class WindowStoreUtils {
private static final int TIMESTAMP_SIZE = 8;
/** Inner byte array serde used for segments */
- public static final Serde<Bytes> INNER_KEY_SERDE = Serdes.Bytes();
- public static final Serde<byte[]> INNER_VALUE_SERDE = Serdes.ByteArray();
- public static final StateSerdes<Bytes, byte[]> INNER_SERDES = new StateSerdes<>("rocksDB-inner", INNER_KEY_SERDE, INNER_VALUE_SERDE);
+ static final Serde<Bytes> INNER_KEY_SERDE = Serdes.Bytes();
+ static final Serde<byte[]> INNER_VALUE_SERDE = Serdes.ByteArray();
+ static final StateSerdes<Bytes, byte[]> INNER_SERDES = new StateSerdes<>("rocksDB-inner", INNER_KEY_SERDE, INNER_VALUE_SERDE);
-
- public static <K> byte[] toBinaryKey(K key, final long timestamp, final int seqnum, StateSerdes<K, ?> serdes) {
+ static <K> Bytes 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) {
+ static Bytes 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);
buf.putInt(seqnum);
- return buf.array();
+ return Bytes.wrap(buf.array());
}
- public static <K> K keyFromBinaryKey(byte[] binaryKey, StateSerdes<K, ?> serdes) {
+ static <K> K keyFromBinaryKey(byte[] binaryKey, StateSerdes<K, ?> serdes) {
byte[] bytes = new byte[binaryKey.length - TIMESTAMP_SIZE - SEQNUM_SIZE];
System.arraycopy(binaryKey, 0, bytes, 0, bytes.length);
@@ -59,7 +58,7 @@ public class WindowStoreUtils {
return serdes.keyFrom(bytes);
}
- public static Bytes bytesKeyFromBinaryKey(byte[] binaryKey) {
+ static Bytes bytesKeyFromBinaryKey(byte[] binaryKey) {
byte[] bytes = new byte[binaryKey.length - TIMESTAMP_SIZE - SEQNUM_SIZE];
System.arraycopy(binaryKey, 0, bytes, 0, bytes.length);
@@ -67,11 +66,11 @@ public class WindowStoreUtils {
return Bytes.wrap(bytes);
}
- public static long timestampFromBinaryKey(byte[] binaryKey) {
+ static long timestampFromBinaryKey(byte[] binaryKey) {
return ByteBuffer.wrap(binaryKey).getLong(binaryKey.length - TIMESTAMP_SIZE - SEQNUM_SIZE);
}
- public static int sequenceNumberFromBinaryKey(byte[] binaryKey) {
+ static int sequenceNumberFromBinaryKey(byte[] binaryKey) {
return ByteBuffer.wrap(binaryKey).getInt(binaryKey.length - SEQNUM_SIZE);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedSessionStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedSessionStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedSessionStoreIterator.java
new file mode 100644
index 0000000..b8db3a6
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedSessionStoreIterator.java
@@ -0,0 +1,89 @@
+/**
+ * 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.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionKeySerde;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.StateSerdes;
+
+class WrappedSessionStoreIterator<K, V> implements KeyValueIterator<Windowed<K>, V> {
+ final KeyValueIterator<Bytes, byte[]> bytesIterator;
+ private final StateSerdes<K, V> serdes;
+
+ // this is optimizing the case when underlying is already a bytes store iterator, in which we can avoid Bytes.wrap() costs
+ private static class WrappedSessionStoreBytesIterator extends WrappedSessionStoreIterator<Bytes, byte[]> {
+ WrappedSessionStoreBytesIterator(final KeyValueIterator<Bytes, byte[]> underlying,
+ final StateSerdes<Bytes, byte[]> serdes) {
+ super(underlying, serdes);
+ }
+
+ @Override
+ public Windowed<Bytes> peekNextKey() {
+ final Bytes key = bytesIterator.peekNextKey();
+
+ return SessionKeySerde.fromBytes(key);
+ }
+
+ @Override
+ public KeyValue<Windowed<Bytes>, byte[]> next() {
+ final KeyValue<Bytes, byte[]> next = bytesIterator.next();
+ return KeyValue.pair(SessionKeySerde.fromBytes(next.key), next.value);
+ }
+ }
+
+ static WrappedSessionStoreIterator<Bytes, byte[]> bytesIterator(final KeyValueIterator<Bytes, byte[]> underlying,
+ final StateSerdes<Bytes, byte[]> serdes) {
+ return new WrappedSessionStoreBytesIterator(underlying, serdes);
+ }
+
+ WrappedSessionStoreIterator(final KeyValueIterator<Bytes, byte[]> bytesIterator, final StateSerdes<K, V> serdes) {
+ this.bytesIterator = bytesIterator;
+ this.serdes = serdes;
+ }
+
+ @Override
+ public void close() {
+ bytesIterator.close();
+ }
+
+ @Override
+ public Windowed<K> peekNextKey() {
+ final Bytes bytes = bytesIterator.peekNextKey();
+ return SessionKeySerde.from(bytes.get(), serdes.keyDeserializer());
+ }
+
+ @Override
+ public boolean hasNext() {
+ return bytesIterator.hasNext();
+ }
+
+ @Override
+ public KeyValue<Windowed<K>, V> next() {
+ final KeyValue<Bytes, byte[]> next = bytesIterator.next();
+ return KeyValue.pair(SessionKeySerde.from(next.key.get(), serdes.keyDeserializer()), serdes.valueFrom(next.value));
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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
index 3d80b98..cbde4e7 100644
--- 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
@@ -34,10 +34,10 @@ interface WrappedStateStore extends StateStore {
*/
StateStore inner();
- abstract class AbstractWrappedStateStore implements WrappedStateStore {
+ abstract class AbstractStateStore implements WrappedStateStore {
final StateStore innerState;
- AbstractWrappedStateStore(StateStore inner) {
+ AbstractStateStore(StateStore inner) {
this.innerState = inner;
}
@@ -85,6 +85,4 @@ interface WrappedStateStore extends StateStore {
innerState.close();
}
}
-
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedWindowStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedWindowStoreIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedWindowStoreIterator.java
new file mode 100644
index 0000000..1e48324
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedWindowStoreIterator.java
@@ -0,0 +1,93 @@
+/**
+ * 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 org.apache.kafka.streams.state.WindowStoreIterator;
+
+import java.util.NoSuchElementException;
+
+class WrappedWindowStoreIterator<V> implements WindowStoreIterator<V> {
+ final KeyValueIterator<Bytes, byte[]> bytesIterator;
+ private final StateSerdes<?, V> serdes;
+
+ // this is optimizing the case when underlying is already a bytes store iterator, in which we can avoid Bytes.wrap() costs
+ private static class WrappedWindowStoreBytesIterator extends WrappedWindowStoreIterator<byte[]> {
+ WrappedWindowStoreBytesIterator(final KeyValueIterator<Bytes, byte[]> underlying,
+ final StateSerdes<Bytes, byte[]> serdes) {
+ super(underlying, serdes);
+ }
+
+ @Override
+ public KeyValue<Long, byte[]> next() {
+ if (!bytesIterator.hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ final KeyValue<Bytes, byte[]> next = bytesIterator.next();
+ final long timestamp = WindowStoreUtils.timestampFromBinaryKey(next.key.get());
+ final byte[] value = next.value;
+ return KeyValue.pair(timestamp, value);
+ }
+ }
+
+ static WrappedWindowStoreIterator<byte[]> bytesIterator(final KeyValueIterator<Bytes, byte[]> underlying,
+ final StateSerdes<Bytes, byte[]> serdes) {
+ return new WrappedWindowStoreBytesIterator(underlying, serdes);
+ }
+
+ WrappedWindowStoreIterator(final KeyValueIterator<Bytes, byte[]> bytesIterator, final StateSerdes<?, V> serdes) {
+ this.bytesIterator = bytesIterator;
+ this.serdes = serdes;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return bytesIterator.hasNext();
+ }
+
+ /**
+ * @throws NoSuchElementException if no next element exists
+ */
+ @Override
+ public KeyValue<Long, V> next() {
+ final KeyValue<Bytes, byte[]> next = bytesIterator.next();
+ final long timestamp = WindowStoreUtils.timestampFromBinaryKey(next.key.get());
+ final V value = serdes.valueFrom(next.value);
+ return KeyValue.pair(timestamp, value);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("remove() is not supported in " + getClass().getName());
+ }
+
+ @Override
+ public void close() {
+ bytesIterator.close();
+ }
+
+ @Override
+ public Long peekNextKey() {
+ return WindowStoreUtils.timestampFromBinaryKey(bytesIterator.peekNextKey().get());
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
index 3a0f490..3be6373 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionKeySerdeTest.java
@@ -17,8 +17,10 @@
package org.apache.kafka.streams.kstream.internals;
+import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.kstream.Window;
import org.apache.kafka.streams.kstream.Windowed;
import org.junit.Test;
@@ -28,60 +30,79 @@ import static org.junit.Assert.assertNull;
public class SessionKeySerdeTest {
+ final private String topic = "topic";
+ final private String key = "key";
+ final private long startTime = 50L;
+ final private long endTime = 100L;
+ final private Window window = new SessionWindow(startTime, endTime);
+ final private Windowed<String> windowedKey = new Windowed<>(key, window);
+ final private Serde<String> serde = Serdes.String();
+ final private SessionKeySerde<String> sessionKeySerde = new SessionKeySerde<>(serde);
+
@Test
public void shouldSerializeDeserialize() throws Exception {
- final Windowed<Long> key = new Windowed<>(1L, new SessionWindow(10, 100));
- final SessionKeySerde<Long> serde = new SessionKeySerde<>(Serdes.Long());
- final byte[] bytes = serde.serializer().serialize("t", key);
- final Windowed<Long> result = serde.deserializer().deserialize("t", bytes);
- assertEquals(key, result);
+ final byte[] bytes = sessionKeySerde.serializer().serialize(topic, windowedKey);
+ final Windowed<String> result = sessionKeySerde.deserializer().deserialize(topic, bytes);
+ assertEquals(windowedKey, result);
}
@Test
public void shouldSerializeNullToNull() throws Exception {
- final SessionKeySerde<String> serde = new SessionKeySerde<>(Serdes.String());
- assertNull(serde.serializer().serialize("t", null));
+ assertNull(sessionKeySerde.serializer().serialize(topic, null));
}
@Test
public void shouldDeSerializeEmtpyByteArrayToNull() throws Exception {
- final SessionKeySerde<String> serde = new SessionKeySerde<>(Serdes.String());
- assertNull(serde.deserializer().deserialize("t", new byte[0]));
+ assertNull(sessionKeySerde.deserializer().deserialize(topic, new byte[0]));
}
@Test
public void shouldDeSerializeNullToNull() throws Exception {
- final SessionKeySerde<String> serde = new SessionKeySerde<>(Serdes.String());
- assertNull(serde.deserializer().deserialize("t", null));
+ assertNull(sessionKeySerde.deserializer().deserialize(topic, null));
}
@Test
public void shouldConvertToBinaryAndBack() throws Exception {
- final Windowed<String> key = new Windowed<>("key", new SessionWindow(10, 20));
- final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
final Windowed<String> result = SessionKeySerde.from(serialized.get(), Serdes.String().deserializer());
- assertEquals(key, result);
+ assertEquals(windowedKey, result);
}
@Test
public void shouldExtractEndTimeFromBinary() throws Exception {
- final Windowed<String> key = new Windowed<>("key", new SessionWindow(10, 100));
- final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
- assertEquals(100, SessionKeySerde.extractEnd(serialized.get()));
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
+ assertEquals(endTime, SessionKeySerde.extractEnd(serialized.get()));
}
@Test
public void shouldExtractStartTimeFromBinary() throws Exception {
- final Windowed<String> key = new Windowed<>("key", new SessionWindow(50, 100));
- final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
- assertEquals(50, SessionKeySerde.extractStart(serialized.get()));
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
+ assertEquals(startTime, SessionKeySerde.extractStart(serialized.get()));
+ }
+
+ @Test
+ public void shouldExtractWindowFromBindary() throws Exception {
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
+ assertEquals(window, SessionKeySerde.extractWindow(serialized.get()));
}
@Test
public void shouldExtractKeyBytesFromBinary() throws Exception {
- final Windowed<String> key = new Windowed<>("blah", new SessionWindow(50, 100));
- final Bytes serialized = SessionKeySerde.toBinary(key, Serdes.String().serializer());
- assertArrayEquals("blah".getBytes(), SessionKeySerde.extractKeyBytes(serialized.get()));
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
+ assertArrayEquals(key.getBytes(), SessionKeySerde.extractKeyBytes(serialized.get()));
}
+ @Test
+ public void shouldExtractKeyFromBinary() throws Exception {
+ final Bytes serialized = SessionKeySerde.toBinary(windowedKey, serde.serializer());
+ assertEquals(windowedKey, SessionKeySerde.from(serialized.get(), serde.deserializer()));
+ }
+
+ @Test
+ public void shouldExtractBytesKeyFromBinary() throws Exception {
+ final Bytes bytesKey = Bytes.wrap(key.getBytes());
+ final Windowed<Bytes> windowedBytesKey = new Windowed<>(bytesKey, window);
+ final Bytes serialized = SessionKeySerde.bytesToBinary(windowedBytesKey);
+ assertEquals(windowedBytesKey, SessionKeySerde.fromBytes(serialized));
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
index ba27c53..ea9e870 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
@@ -232,6 +232,7 @@ public class ProcessorStateManagerTest {
put(nonPersistentStoreName, nonPersistentStoreName);
}
});
+
try {
stateMgr.register(mockStateStore, true, mockStateStore.stateRestoreCallback);
} finally {
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
index a00526f..dc510c5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
@@ -28,7 +28,6 @@ import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.KeyValueIterator;
-import org.apache.kafka.test.InMemoryKeyValueStore;
import org.apache.kafka.test.MockProcessorContext;
import org.junit.Before;
import org.junit.Test;
@@ -58,7 +57,7 @@ public class CachingKeyValueStoreTest {
@Before
public void setUp() throws Exception {
final String storeName = "store";
- underlyingStore = new InMemoryKeyValueStore<>(storeName);
+ underlyingStore = new InMemoryKeyValueStore<>(storeName, Serdes.Bytes(), Serdes.ByteArray());
cacheFlushListener = new CacheFlushListenerStub<>();
store = new CachingKeyValueStore<>(underlyingStore, Serdes.String(), Serdes.String());
store.setFlushListener(cacheFlushListener);
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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 2728aa0..39947b9 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
@@ -46,19 +46,19 @@ import static org.junit.Assert.assertNull;
public class CachingWindowStoreTest {
private static final int MAX_CACHE_SIZE_BYTES = 150;
+ private static final long DEFAULT_TIMESTAMP = 10L;
private static final Long WINDOW_SIZE = 10000L;
private RocksDBSegmentedBytesStore underlying;
private CachingWindowStore<String, String> cachingStore;
private CachingKeyValueStoreTest.CacheFlushListenerStub<Windowed<String>> cacheListener;
private ThreadCache cache;
private String topic;
- private static final long DEFAULT_TIMESTAMP = 10L;
- private WindowStoreKeySchema keySchema;
+ private WindowKeySchema keySchema;
private RocksDBWindowStore<Bytes, byte[]> windowStore;
@Before
public void setUp() throws Exception {
- keySchema = new WindowStoreKeySchema();
+ keySchema = new WindowKeySchema();
underlying = new RocksDBSegmentedBytesStore("test", 30000, 3, keySchema);
windowStore = new RocksDBWindowStore<>(underlying, Serdes.Bytes(), Serdes.ByteArray(), false);
cacheListener = new CachingKeyValueStoreTest.CacheFlushListenerStub<>();
@@ -152,7 +152,7 @@ public class CachingWindowStoreTest {
@Test
public void shouldIterateCacheAndStore() throws Exception {
final Bytes key = Bytes.wrap("1" .getBytes());
- underlying.put(Bytes.wrap(WindowStoreUtils.toBinaryKey(key, DEFAULT_TIMESTAMP, 0, WindowStoreUtils.INNER_SERDES)), "a".getBytes());
+ underlying.put(WindowStoreUtils.toBinaryKey(key, DEFAULT_TIMESTAMP, 0, WindowStoreUtils.INNER_SERDES), "a".getBytes());
cachingStore.put("1", "b", DEFAULT_TIMESTAMP + WINDOW_SIZE);
final WindowStoreIterator<String> fetch = cachingStore.fetch("1", DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE);
assertEquals(KeyValue.pair(DEFAULT_TIMESTAMP, "a"), fetch.next());
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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
index 58b42e3..123d8ff 100644
--- 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
@@ -22,7 +22,6 @@ 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;
@@ -40,7 +39,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
public class ChangeLoggingKeyValueBytesStoreTest {
- private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv");
+ private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv", Serdes.Bytes(), Serdes.ByteArray());
private final ChangeLoggingKeyValueBytesStore store = new ChangeLoggingKeyValueBytesStore(inner);
private final Map sent = new HashMap<>();
private final Bytes hi = Bytes.wrap("hi".getBytes());
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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
index d45f82e..27503ea 100644
--- 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
@@ -24,7 +24,6 @@ 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;
@@ -43,7 +42,7 @@ import static org.junit.Assert.assertFalse;
public class ChangeLoggingKeyValueStoreTest {
- private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv");
+ private final InMemoryKeyValueStore<Bytes, byte[]> inner = new InMemoryKeyValueStore<>("kv", Serdes.Bytes(), Serdes.ByteArray());
private final Serde<String> keySerde = Serdes.String();
private final Serde<String> valueSerde = Serdes.String();
private final ChangeLoggingKeyValueStore<String, String> store
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
index 0fd6001..763ca48 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java
@@ -98,7 +98,7 @@ public class CompositeReadOnlyKeyValueStoreTest {
stubOneUnderlying.put("b", "b");
stubOneUnderlying.put("c", "c");
- final List<KeyValue<String, String>> results = toList(theStore.range("a", "c"));
+ final List<KeyValue<String, String>> results = toList(theStore.range("a", "b"));
assertTrue(results.contains(new KeyValue<>("a", "a")));
assertTrue(results.contains(new KeyValue<>("b", "b")));
assertEquals(2, results.size());
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/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 0ebdd5c..3f5ea6d 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,8 +17,8 @@
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.test.InMemoryKeyValueStore;
import org.junit.Before;
import org.junit.Test;
@@ -34,7 +34,7 @@ public class DelegatingPeekingKeyValueIteratorTest {
@Before
public void setUp() throws Exception {
- store = new InMemoryKeyValueStore<>(name);
+ store = new InMemoryKeyValueStore<>(name, Serdes.String(), Serdes.String());
}
@Test
http://git-wip-us.apache.org/repos/asf/kafka/blob/7ebc5da6/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
index fbcefab..161b791 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MergedSortedCacheKeyValueStoreIteratorTest.java
@@ -24,7 +24,6 @@ import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StateSerdes;
-import org.apache.kafka.test.InMemoryKeyValueStore;
import org.junit.Before;
import org.junit.Test;
@@ -40,7 +39,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Before
public void setUp() throws Exception {
- store = new InMemoryKeyValueStore<>(namespace);
+ store = new InMemoryKeyValueStore<>(namespace, Serdes.Bytes(), Serdes.ByteArray());
cache = new ThreadCache("testCache", 10000L, new MockStreamsMetrics(new Metrics()));
}
@@ -49,13 +48,13 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}, {11}};
for (int i = 0; i < bytes.length; i += 2) {
store.put(Bytes.wrap(bytes[i]), bytes[i]);
- cache.put(namespace, bytes[i + 1], new LRUCacheEntry(bytes[i + 1]));
+ cache.put(namespace, Bytes.wrap(bytes[i + 1]), new LRUCacheEntry(bytes[i + 1]));
}
final Bytes from = Bytes.wrap(new byte[]{2});
final Bytes to = Bytes.wrap(new byte[]{9});
- final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("store", store.range(from, to));
- final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, from.get(), to.get());
+ final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("store", store.range(from, to));
+ final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, from, to);
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator = new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes);
byte[][] values = new byte[8][];
@@ -73,7 +72,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
public void shouldSkipLargerDeletedCacheValue() throws Exception {
final byte[][] bytes = {{0}, {1}};
store.put(Bytes.wrap(bytes[0]), bytes[0]);
- cache.put(namespace, bytes[1], new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[1]), new LRUCacheEntry(null));
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator = createIterator();
assertArrayEquals(bytes[0], iterator.next().key);
assertFalse(iterator.hasNext());
@@ -82,7 +81,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Test
public void shouldSkipSmallerDeletedCachedValue() throws Exception {
final byte[][] bytes = {{0}, {1}};
- cache.put(namespace, bytes[0], new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[0]), new LRUCacheEntry(null));
store.put(Bytes.wrap(bytes[1]), bytes[1]);
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator = createIterator();
assertArrayEquals(bytes[1], iterator.next().key);
@@ -92,7 +91,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Test
public void shouldIgnoreIfDeletedInCacheButExistsInStore() throws Exception {
final byte[][] bytes = {{0}};
- cache.put(namespace, bytes[0], new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[0]), new LRUCacheEntry(null));
store.put(Bytes.wrap(bytes[0]), bytes[0]);
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator = createIterator();
assertFalse(iterator.hasNext());
@@ -102,8 +101,9 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
public void shouldNotHaveNextIfAllCachedItemsDeleted() throws Exception {
final byte[][] bytes = {{0}, {1}, {2}};
for (byte[] aByte : bytes) {
- store.put(Bytes.wrap(aByte), aByte);
- cache.put(namespace, aByte, new LRUCacheEntry(null));
+ Bytes aBytes = Bytes.wrap(aByte);
+ store.put(aBytes, aByte);
+ cache.put(namespace, aBytes, new LRUCacheEntry(null));
}
assertFalse(createIterator().hasNext());
}
@@ -112,7 +112,7 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
public void shouldNotHaveNextIfOnlyCacheItemsAndAllDeleted() throws Exception {
final byte[][] bytes = {{0}, {1}, {2}};
for (byte[] aByte : bytes) {
- cache.put(namespace, aByte, new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(aByte), new LRUCacheEntry(null));
}
assertFalse(createIterator().hasNext());
}
@@ -121,14 +121,15 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
public void shouldSkipAllDeletedFromCache() throws Exception {
final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}, {11}};
for (byte[] aByte : bytes) {
- store.put(Bytes.wrap(aByte), aByte);
- cache.put(namespace, aByte, new LRUCacheEntry(aByte));
+ Bytes aBytes = Bytes.wrap(aByte);
+ store.put(aBytes, aByte);
+ cache.put(namespace, aBytes, new LRUCacheEntry(aByte));
}
- cache.put(namespace, bytes[1], new LRUCacheEntry(null));
- cache.put(namespace, bytes[2], new LRUCacheEntry(null));
- cache.put(namespace, bytes[3], new LRUCacheEntry(null));
- cache.put(namespace, bytes[8], new LRUCacheEntry(null));
- cache.put(namespace, bytes[11], new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[1]), new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[2]), new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[3]), new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[8]), new LRUCacheEntry(null));
+ cache.put(namespace, Bytes.wrap(bytes[11]), new LRUCacheEntry(null));
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator = createIterator();
assertArrayEquals(bytes[0], iterator.next().key);
@@ -144,19 +145,19 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
@Test
public void shouldPeekNextKey() throws Exception {
- final KeyValueStore<Bytes, byte[]> kv = new InMemoryKeyValueStore<>("one");
+ final KeyValueStore<Bytes, byte[]> kv = new InMemoryKeyValueStore<>("one", Serdes.Bytes(), Serdes.ByteArray());
final ThreadCache cache = new ThreadCache("testCache", 1000000L, new MockStreamsMetrics(new Metrics()));
byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}};
final String namespace = "one";
for (int i = 0; i < bytes.length - 1; i += 2) {
kv.put(Bytes.wrap(bytes[i]), bytes[i]);
- cache.put(namespace, bytes[i + 1], new LRUCacheEntry(bytes[i + 1]));
+ cache.put(namespace, Bytes.wrap(bytes[i + 1]), new LRUCacheEntry(bytes[i + 1]));
}
final Bytes from = Bytes.wrap(new byte[]{2});
final Bytes to = Bytes.wrap(new byte[]{9});
final KeyValueIterator<Bytes, byte[]> storeIterator = kv.range(from, to);
- final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, from.get(), to.get());
+ final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(namespace, from, to);
final MergedSortedCacheKeyValueStoreIterator<byte[], byte[]> iterator =
new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator,
@@ -178,6 +179,4 @@ public class MergedSortedCacheKeyValueStoreIteratorTest {
final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>("store", store.all());
return new MergedSortedCacheKeyValueStoreIterator<>(cacheIterator, storeIterator, serdes);
}
-
-
}
\ No newline at end of file