From 129ece168887d189f9a2c19eeb06c334ed1f2cb9 Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Thu, 15 Oct 2015 13:50:28 -0500 Subject: [PATCH] KAFKA-2594 Added InMemoryLRUCacheStore Added a new KeyValueStore implementation that keeps a maximum number of entries in-memory, and as the size exceeds the capacity the least-recently used entry is removed from the store and the backing topic. Also added unit tests for this new store. --- .../streams/state/InMemoryLRUCacheStore.java | 180 ++++++++++++++++++ .../streams/state/MeteredKeyValueStore.java | 44 ++++- .../apache/kafka/streams/state/Stores.java | 25 ++- .../state/InMemoryLRUCacheStoreTest.java | 148 ++++++++++++++ 4 files changed, 386 insertions(+), 11 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java new file mode 100644 index 0000000000000..1b96c59382c88 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java @@ -0,0 +1,180 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.state; + +import org.apache.kafka.common.utils.SystemTime; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streams.processor.ProcessorContext; + +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.NavigableSet; +import java.util.TreeSet; + +/** + * An in-memory key-value store that is limited in size and retains a maximum number of most recently used entries. + * + * @param The key type + * @param The value type + * + */ +public class InMemoryLRUCacheStore extends MeteredKeyValueStore { + + protected static InMemoryLRUCacheStore create(String name, int capacity, ProcessorContext context, + Serdes serdes, Time time) { + if (time == null) time = new SystemTime(); + MemoryLRUCache cache = new MemoryLRUCache(name, capacity); + final InMemoryLRUCacheStore store = new InMemoryLRUCacheStore<>(name, context, cache, serdes, time); + cache.whenEldestRemoved(new EldestEntryRemovalListener() { + @Override + public void apply(K key, V value) { + store.removed(key); + } + }); + return store; + + } + + private InMemoryLRUCacheStore(String name, ProcessorContext context, MemoryLRUCache cache, Serdes serdes, Time time) { + super(name, cache, context, serdes, "kafka-streams", time); + } + + private static interface EldestEntryRemovalListener { + public void apply(K key, V value); + } + + protected static final class MemoryLRUCache implements KeyValueStore { + + private final String name; + private final Map map; + private final NavigableSet keys; + private EldestEntryRemovalListener listener; + + public MemoryLRUCache(String name, final int maxCacheSize) { + this.name = name; + this.keys = new TreeSet<>(); + // leave room for one extra entry to handle adding an entry before the oldest can be removed + this.map = new LinkedHashMap(maxCacheSize + 1, 1.01f, true) { + private static final long serialVersionUID = 1L; + + @Override + protected boolean removeEldestEntry(Map.Entry eldest) { + if (size() > maxCacheSize) { + K key = eldest.getKey(); + keys.remove(key); + if (listener != null) listener.apply(key, eldest.getValue()); + return true; + } + return false; + } + }; + } + + protected void whenEldestRemoved(EldestEntryRemovalListener listener) { + this.listener = listener; + } + + @Override + public String name() { + return this.name; + } + + @Override + public boolean persistent() { + return false; + } + + @Override + public V get(K key) { + return this.map.get(key); + } + + @Override + public void put(K key, V value) { + this.map.put(key, value); + this.keys.add(key); + } + + @Override + public void putAll(List> entries) { + for (Entry entry : entries) + put(entry.key(), entry.value()); + } + + @Override + public V delete(K key) { + V value = this.map.remove(key); + this.keys.remove(key); + return value; + } + + @Override + public KeyValueIterator range(K from, K to) { + return new MemoryLRUCache.CacheIterator(this.keys.subSet(from, true, to, false).iterator(), this.map); + } + + @Override + public KeyValueIterator all() { + return new MemoryLRUCache.CacheIterator(this.keys.iterator(), this.map); + } + + @Override + public void flush() { + // do-nothing since it is in-memory + } + + @Override + public void close() { + // do-nothing + } + + private static class CacheIterator implements KeyValueIterator { + private final Iterator keys; + private final Map entries; + private K lastKey; + + public CacheIterator(Iterator keys, Map entries) { + this.keys = keys; + this.entries = entries; + } + + @Override + public boolean hasNext() { + return keys.hasNext(); + } + + @Override + public Entry next() { + lastKey = keys.next(); + return new Entry<>(lastKey, entries.get(lastKey)); + } + + @Override + public void remove() { + keys.remove(); + entries.remove(lastKey); + } + + @Override + public void close() { + // do nothing + } + } + } +} \ No newline at end of file diff --git a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java index 90eee054371cd..9a652ac91bef5 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java @@ -50,7 +50,9 @@ public class MeteredKeyValueStore implements KeyValueStore { private final String topic; private final int partition; private final Set dirty; + private final Set removed; private final int maxDirty; + private final int maxRemoved; private final ProcessorContext context; // always wrap the logged store with the metered store @@ -76,7 +78,9 @@ public MeteredKeyValueStore(final String name, final KeyValueStore inner, this.context = context; this.dirty = new HashSet(); - this.maxDirty = 100; // TODO: this needs to be configurable + this.removed = new HashSet(); + this.maxDirty = 100; // TODO: this needs to be configurable + this.maxRemoved = 100; // TODO: this needs to be configurable // register and possibly restore the state from the logs long startNs = time.nanoseconds(); @@ -123,8 +127,8 @@ public void put(K key, V value) { this.inner.put(key, value); this.dirty.add(key); - if (this.dirty.size() > this.maxDirty) - logChange(); + this.removed.remove(key); + maybeLogChange(); } finally { this.metrics.recordLatency(this.putTime, startNs, time.nanoseconds()); } @@ -137,11 +141,12 @@ public void putAll(List> entries) { this.inner.putAll(entries); for (Entry entry : entries) { - this.dirty.add(entry.key()); + K key = entry.key(); + this.dirty.add(key); + this.removed.remove(key); } - if (this.dirty.size() > this.maxDirty) - logChange(); + maybeLogChange(); } finally { this.metrics.recordLatency(this.putAllTime, startNs, time.nanoseconds()); } @@ -153,9 +158,9 @@ public V delete(K key) { try { V value = this.inner.delete(key); - this.dirty.add(key); - if (this.dirty.size() > this.maxDirty) - logChange(); + this.dirty.remove(key); + this.removed.add(key); + maybeLogChange(); return value; } finally { @@ -163,6 +168,18 @@ public V delete(K key) { } } + /** + * Called when the underlying {@link #inner} {@link KeyValueStore} removes an entry in response to a call from this + * store other than {@link #delete(Object)}. + * + * @param key the key for the entry that the inner store removed + */ + protected void removed(K key) { + this.dirty.remove(key); + this.removed.add(key); + maybeLogChange(); + } + @Override public KeyValueIterator range(K from, K to) { return new MeteredKeyValueIterator(this.inner.range(from, to), this.rangeTime); @@ -189,16 +206,25 @@ public void flush() { } } + private void maybeLogChange() { + if (this.dirty.size() > this.maxDirty || this.removed.size() > this.maxRemoved) + logChange(); + } + private void logChange() { RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector(); if (collector != null) { Serializer keySerializer = serialization.keySerializer(); Serializer valueSerializer = serialization.valueSerializer(); + for (K k : this.removed) { + collector.send(new ProducerRecord<>(this.topic, this.partition, k, (V) null), keySerializer, valueSerializer); + } for (K k : this.dirty) { V v = this.inner.get(k); collector.send(new ProducerRecord<>(this.topic, this.partition, k, v), keySerializer, valueSerializer); } + this.removed.clear(); this.dirty.clear(); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java index 171ed44c76727..8ec73fdb656c6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java @@ -54,8 +54,20 @@ public KeyValueFactory withValues(final Serializer valueSerializer, @Override public InMemoryKeyValueFactory inMemory() { return new InMemoryKeyValueFactory() { + private int capacity = Integer.MAX_VALUE; + + @Override + public InMemoryKeyValueFactory maxEntries(int capacity) { + if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive"); + this.capacity = capacity; + return this; + } + @Override public KeyValueStore build() { + if (capacity < Integer.MAX_VALUE) { + return InMemoryLRUCacheStore.create(name, capacity, context, serdes, null); + } return new InMemoryKeyValueStore<>(name, context, serdes, null); } }; @@ -138,7 +150,7 @@ public ValueFactory withKeys(Class keyClass) { } /** - * The interface used to specify the type of values for key-value stores. + * The factory for creating off-heap key-value stores. * * @param the type of keys */ @@ -200,7 +212,6 @@ public KeyValueFactory withValues(Class valueClass) { * @return the interface used to specify the remaining key-value store options; never null */ public abstract KeyValueFactory withValues(Serializer valueSerializer, Deserializer valueDeserializer); - } /** @@ -234,6 +245,16 @@ public static interface KeyValueFactory { * @param the type of values */ public static interface InMemoryKeyValueFactory { + /** + * Limits the in-memory key-value store to hold a maximum number of entries. The default is {@link Integer#MAX_VALUE}, which is + * equivalent to not placing a limit on the number of entries. + * + * @param capacity the maximum capacity of the in-memory cache; should be one less than a power of 2 + * @return this factory + * @throws IllegalArgumentException if the capacity is not positive + */ + InMemoryKeyValueFactory maxEntries(int capacity); + /** * Return the new key-value store. * @return the key-value store; never null diff --git a/streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java new file mode 100644 index 0000000000000..6b96d3ada9142 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java @@ -0,0 +1,148 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.state; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; +import org.junit.Test; + +public class InMemoryLRUCacheStoreTest { + + @Test + public void testPutGetRange() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(); + KeyValueStore store = Stores.create("my-store", driver.context()) + .withIntegerKeys().withStringValues() + .inMemory().maxEntries(3) + .build(); + + // Verify that the store reads and writes correctly, keeping only the last 2 entries ... + store.put(0, "zero"); + store.put(1, "one"); + store.put(2, "two"); + store.put(3, "three"); + store.put(4, "four"); + store.put(5, "five"); + + // It should only keep the last 4 added ... + assertEquals(3, driver.sizeOf(store)); + assertNull(store.get(0)); + assertNull(store.get(1)); + assertNull(store.get(2)); + assertEquals("three", store.get(3)); + assertEquals("four", store.get(4)); + assertEquals("five", store.get(5)); + store.delete(5); + + // Flush the store and verify all current entries were properly flushed ... + store.flush(); + assertNull(driver.flushedEntryStored(0)); + assertNull(driver.flushedEntryStored(1)); + assertNull(driver.flushedEntryStored(2)); + assertEquals("three", driver.flushedEntryStored(3)); + assertEquals("four", driver.flushedEntryStored(4)); + assertNull(driver.flushedEntryStored(5)); + + assertEquals(true, driver.flushedEntryRemoved(0)); + assertEquals(true, driver.flushedEntryRemoved(1)); + assertEquals(true, driver.flushedEntryRemoved(2)); + assertEquals(false, driver.flushedEntryRemoved(3)); + assertEquals(false, driver.flushedEntryRemoved(4)); + assertEquals(true, driver.flushedEntryRemoved(5)); + } + + @SuppressWarnings("unchecked") + @Test + public void testPutGetRangeWithDefaultSerdes() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(); + + Serializer keySer = (Serializer) driver.context().keySerializer(); + Deserializer keyDeser = (Deserializer) driver.context().keyDeserializer(); + Serializer valSer = (Serializer) driver.context().valueSerializer(); + Deserializer valDeser = (Deserializer) driver.context().valueDeserializer(); + KeyValueStore store = Stores.create("my-store", driver.context()) + .withKeys(keySer, keyDeser) + .withValues(valSer, valDeser) + .inMemory().maxEntries(3) + .build(); + + // Verify that the store reads and writes correctly, keeping only the last 2 entries ... + store.put(0, "zero"); + store.put(1, "one"); + store.put(2, "two"); + store.put(3, "three"); + store.put(4, "four"); + store.put(5, "five"); + + // It should only keep the last 4 added ... + assertEquals(3, driver.sizeOf(store)); + assertNull(store.get(0)); + assertNull(store.get(1)); + assertNull(store.get(2)); + assertEquals("three", store.get(3)); + assertEquals("four", store.get(4)); + assertEquals("five", store.get(5)); + store.delete(5); + + // Flush the store and verify all current entries were properly flushed ... + store.flush(); + assertNull(driver.flushedEntryStored(0)); + assertNull(driver.flushedEntryStored(1)); + assertNull(driver.flushedEntryStored(2)); + assertEquals("three", driver.flushedEntryStored(3)); + assertEquals("four", driver.flushedEntryStored(4)); + assertNull(driver.flushedEntryStored(5)); + + assertEquals(true, driver.flushedEntryRemoved(0)); + assertEquals(true, driver.flushedEntryRemoved(1)); + assertEquals(true, driver.flushedEntryRemoved(2)); + assertEquals(false, driver.flushedEntryRemoved(3)); + assertEquals(false, driver.flushedEntryRemoved(4)); + assertEquals(true, driver.flushedEntryRemoved(5)); + } + + @Test + public void testRestore() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(Integer.class, String.class); + + // Add any entries that will be restored to any store + // that uses the driver's context ... + driver.addEntryToRestoreLog(1, "one"); + driver.addEntryToRestoreLog(2, "two"); + driver.addEntryToRestoreLog(4, "four"); + + // Create the store, which should register with the context and automatically + // receive the restore entries ... + KeyValueStore store = Stores.create("my-store", driver.context()) + .withIntegerKeys().withStringValues() + .inMemory().maxEntries(3) + .build(); + + // Verify that the store's contents were properly restored ... + assertEquals(0, driver.checkForRestoredEntries(store)); + + // and there are no other entries ... + assertEquals(3, driver.sizeOf(store)); + } + +} \ No newline at end of file