Skip to content

KAFKA-2594 Added InMemoryLRUCacheStore #256

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rhauch wants to merge 1 commit into from
Closed

KAFKA-2594 Added InMemoryLRUCacheStore #256

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 180 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 <K> The key type
* @param <V> The value type
*
*/
public class InMemoryLRUCacheStore<K, V> extends MeteredKeyValueStore<K, V> {

protected static <K, V> InMemoryLRUCacheStore<K, V> create(String name, int capacity, ProcessorContext context,
Serdes<K, V> serdes, Time time) {
if (time == null) time = new SystemTime();
MemoryLRUCache<K, V> cache = new MemoryLRUCache<K, V>(name, capacity);
final InMemoryLRUCacheStore<K, V> store = new InMemoryLRUCacheStore<>(name, context, cache, serdes, time);
cache.whenEldestRemoved(new EldestEntryRemovalListener<K, V>() {
@Override
public void apply(K key, V value) {
store.removed(key);
}
});
return store;

}

private InMemoryLRUCacheStore(String name, ProcessorContext context, MemoryLRUCache<K, V> cache, Serdes<K, V> serdes, Time time) {
super(name, cache, context, serdes, "kafka-streams", time);
}

private static interface EldestEntryRemovalListener<K, V> {
public void apply(K key, V value);
}

protected static final class MemoryLRUCache<K, V> implements KeyValueStore<K, V> {

private final String name;
private final Map<K, V> map;
private final NavigableSet<K> keys;
private EldestEntryRemovalListener<K, V> 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<K, V>(maxCacheSize + 1, 1.01f, true) {
private static final long serialVersionUID = 1L;

@Override
protected boolean removeEldestEntry(Map.Entry<K, V> 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<K, V> 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<Entry<K, V>> entries) {
for (Entry<K, V> 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<K, V> range(K from, K to) {
return new MemoryLRUCache.CacheIterator<K, V>(this.keys.subSet(from, true, to, false).iterator(), this.map);
}

@Override
public KeyValueIterator<K, V> all() {
return new MemoryLRUCache.CacheIterator<K, V>(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<K, V> implements KeyValueIterator<K, V> {
private final Iterator<K> keys;
private final Map<K, V> entries;
private K lastKey;

public CacheIterator(Iterator<K> keys, Map<K, V> entries) {
this.keys = keys;
this.entries = entries;
}

@Override
public boolean hasNext() {
return keys.hasNext();
}

@Override
public Entry<K, V> 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
}
}
}
}
44 changes: 35 additions & 9 deletions streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,9 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K, V> {
private final String topic;
private final int partition;
private final Set<K> dirty;
private final Set<K> removed;
private final int maxDirty;
private final int maxRemoved;
private final ProcessorContext context;

// always wrap the logged store with the metered store
Expand All @@ -76,7 +78,9 @@ public MeteredKeyValueStore(final String name, final KeyValueStore<K, V> inner,
this.context = context;

this.dirty = new HashSet<K>();
this.maxDirty = 100; // TODO: this needs to be configurable
this.removed = new HashSet<K>();
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();
Expand Down Expand Up @@ -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());
}
Expand All @@ -137,11 +141,12 @@ public void putAll(List<Entry<K, V>> entries) {
this.inner.putAll(entries);

for (Entry<K, V> 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());
}
Expand All @@ -153,16 +158,28 @@ 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 {
this.metrics.recordLatency(this.deleteTime, startNs, time.nanoseconds());
}
}

/**
* 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<K, V> range(K from, K to) {
return new MeteredKeyValueIterator<K, V>(this.inner.range(from, to), this.rangeTime);
Expand All @@ -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<K> keySerializer = serialization.keySerializer();
Serializer<V> 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();
}
}
Expand Down
25 changes: 23 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/state/Stores.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,20 @@ public <V> KeyValueFactory<K, V> withValues(final Serializer<V> valueSerializer,
@Override
public InMemoryKeyValueFactory<K, V> inMemory() {
return new InMemoryKeyValueFactory<K, V>() {
private int capacity = Integer.MAX_VALUE;

@Override
public InMemoryKeyValueFactory<K, V> maxEntries(int capacity) {
if (capacity < 1) throw new IllegalArgumentException("The capacity must be positive");
this.capacity = capacity;
return this;
}

@Override
public KeyValueStore<K, V> build() {
if (capacity < Integer.MAX_VALUE) {
return InMemoryLRUCacheStore.create(name, capacity, context, serdes, null);
}
return new InMemoryKeyValueStore<>(name, context, serdes, null);
}
};
Expand Down Expand Up @@ -138,7 +150,7 @@ public <K> ValueFactory<K> withKeys(Class<K> keyClass) {
}

/**
* The interface used to specify the type of values for key-value stores.
* The factory for creating off-heap key-value stores.
*
* @param <K> the type of keys
*/
Expand Down Expand Up @@ -200,7 +212,6 @@ public <V> KeyValueFactory<K, V> withValues(Class<V> valueClass) {
* @return the interface used to specify the remaining key-value store options; never null
*/
public abstract <V> KeyValueFactory<K, V> withValues(Serializer<V> valueSerializer, Deserializer<V> valueDeserializer);

}

/**
Expand Down Expand Up @@ -234,6 +245,16 @@ public static interface KeyValueFactory<K, V> {
* @param <V> the type of values
*/
public static interface InMemoryKeyValueFactory<K, V> {
/**
* 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<K, V> maxEntries(int capacity);

/**
* Return the new key-value store.
* @return the key-value store; never null
Expand Down
Loading