From dd25a810dad7b1bef27e9ae4a48de66a8e3ced2e Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Wed, 23 Sep 2015 14:05:56 -0500 Subject: [PATCH 1/5] Parameterized RocksDBKeyValueStore --- .../streams/state/RocksDBKeyValueStore.java | 126 ++++++++++++------ 1 file changed, 88 insertions(+), 38 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java index e0962a206ad71..aac5a96b6901a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java @@ -17,11 +17,12 @@ package org.apache.kafka.streams.state; -import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.utils.SystemTime; - import org.apache.kafka.common.utils.Time; +import org.apache.kafka.streams.processor.ProcessorContext; import org.rocksdb.BlockBasedTableConfig; import org.rocksdb.CompactionStyle; import org.rocksdb.CompressionType; @@ -37,17 +38,27 @@ import java.util.List; import java.util.NoSuchElementException; -public class RocksDBKeyValueStore extends MeteredKeyValueStore { +/** + * A {@link KeyValueStore} that stores all entries in a local RocksDB database. + * + * @param the type of keys + * @param the type of values + */ +public class RocksDBKeyValueStore extends MeteredKeyValueStore { - public RocksDBKeyValueStore(String name, ProcessorContext context) { - this(name, context, new SystemTime()); + public RocksDBKeyValueStore(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { + this(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime()); } - public RocksDBKeyValueStore(String name, ProcessorContext context, Time time) { - super(name, new RocksDBStore(name, context), context, "kafka-streams", time); + public RocksDBKeyValueStore(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, Time time) { + super(name, new RocksDBStore(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer), context, "kafka-streams", time); } - private static class RocksDBStore implements KeyValueStore { + private static class RocksDBStore implements KeyValueStore { private static final int TTL_NOT_USED = -1; @@ -61,6 +72,11 @@ private static class RocksDBStore implements KeyValueStore { private static final CompactionStyle COMPACTION_STYLE = CompactionStyle.UNIVERSAL; private static final String DB_FILE_DIR = "rocksdb"; + private final Serializer keySerializer; + private final Serializer valueSerializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; + private final String topic; private final int partition; private final ProcessorContext context; @@ -75,10 +91,15 @@ private static class RocksDBStore implements KeyValueStore { private RocksDB db; @SuppressWarnings("unchecked") - public RocksDBStore(String name, ProcessorContext context) { + public RocksDBStore(String name, ProcessorContext context, Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { this.topic = name; this.partition = context.id(); this.context = context; + this.keySerializer = keySerializer != null ? keySerializer : (Serializer)context.keySerializer(); + this.keyDeserializer = keyDeserializer != null ? keyDeserializer : (Deserializer)context.keyDeserializer(); + this.valueSerializer = valueSerializer != null ? valueSerializer : (Serializer)context.valueSerializer(); + this.valueDeserializer = valueDeserializer != null ? valueDeserializer : (Deserializer)context.valueDeserializer(); // initialize the rocksdb options BlockBasedTableConfig tableConfig = new BlockBasedTableConfig(); @@ -130,24 +151,24 @@ public String name() { public boolean persistent() { return false; } - + @Override - public byte[] get(byte[] key) { + public V get(K key) { try { - return this.db.get(key); + return valueFrom(this.db.get(rawKey(key))); } catch (RocksDBException e) { // TODO: this needs to be handled more accurately throw new KafkaException("Error while executing get " + key.toString() + " from store " + this.topic, e); } } - + @Override - public void put(byte[] key, byte[] value) { + public void put(K key, V value) { try { if (value == null) { - db.remove(wOptions, key); + db.remove(wOptions, rawKey(key)); } else { - db.put(wOptions, key, value); + db.put(wOptions, rawKey(key), rawValue(value)); } } catch (RocksDBException e) { // TODO: this needs to be handled more accurately @@ -156,28 +177,28 @@ public void put(byte[] key, byte[] value) { } @Override - public void putAll(List> entries) { - for (Entry entry : entries) + public void putAll(List> entries) { + for (Entry entry : entries) put(entry.key(), entry.value()); } - + @Override - public byte[] delete(byte[] key) { - byte[] value = get(key); + public V delete(K key) { + V value = get(key); put(key, null); return value; } @Override - public KeyValueIterator range(byte[] from, byte[] to) { - return new RocksDBRangeIterator(db.newIterator(), from, to); + public KeyValueIterator range(K from, K to) { + return new RocksDBRangeIterator(db.newIterator(), topic, keySerializer , keyDeserializer, valueDeserializer, from, to); } @Override - public KeyValueIterator all() { + public KeyValueIterator all() { RocksIterator innerIter = db.newIterator(); innerIter.seekToFirst(); - return new RocksDbIterator(innerIter); + return new RocksDbIterator(innerIter, topic, keyDeserializer, valueDeserializer); } @Override @@ -196,19 +217,46 @@ public void close() { db.close(); } - private static class RocksDbIterator implements KeyValueIterator { + protected final V valueFrom( byte[] rawValue ) { + return valueDeserializer.deserialize(topic, rawValue); + } + + protected final byte[] rawKey( K key ) { + return keySerializer.serialize(topic, key); + } + + protected final byte[] rawValue( V value ) { + return valueSerializer.serialize(topic, value); + } + + private static class RocksDbIterator implements KeyValueIterator { private final RocksIterator iter; + private final String topic; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; - public RocksDbIterator(RocksIterator iter) { + public RocksDbIterator(RocksIterator iter, String topic, + Deserializer keyDeserializer, Deserializer valueDeserializer) { this.iter = iter; + this.topic = topic; + this.keyDeserializer = keyDeserializer; + this.valueDeserializer = valueDeserializer; } protected byte[] peekKey() { - return this.getEntry().key(); + return iter.key(); + } + + protected final K keyFrom( byte[] rawKey ) { + return keyDeserializer.deserialize(topic, rawKey); + } + + protected final V valueFrom( byte[] rawValue ) { + return valueDeserializer.deserialize(topic, rawValue); } - protected Entry getEntry() { - return new Entry<>(iter.key(), iter.value()); + protected Entry getEntry() { + return new Entry<>(keyFrom(iter.key()),valueFrom(iter.value())); } @Override @@ -217,13 +265,12 @@ public boolean hasNext() { } @Override - public Entry next() { + public Entry next() { if (!hasNext()) throw new NoSuchElementException(); - Entry entry = this.getEntry(); + Entry entry = this.getEntry(); iter.next(); - return entry; } @@ -253,17 +300,20 @@ public int compare(byte[] left, byte[] right) { } } - private static class RocksDBRangeIterator extends RocksDbIterator { + private static class RocksDBRangeIterator extends RocksDbIterator { // RocksDB's JNI interface does not expose getters/setters that allow the // comparator to be pluggable, and the default is lexicographic, so it's // safe to just force lexicographic comparator here for now. private final Comparator comparator = new LexicographicComparator(); byte[] to; - public RocksDBRangeIterator(RocksIterator iter, byte[] from, byte[] to) { - super(iter); - iter.seek(from); - this.to = to; + public RocksDBRangeIterator(RocksIterator iter, String topic, + Serializer keySerializer, Deserializer keyDeserializer, + Deserializer valueDeserializer, + K from, K to) { + super(iter,topic,keyDeserializer,valueDeserializer); + iter.seek(keySerializer.serialize(topic, from)); + this.to = keySerializer.serialize(topic, to); } @Override From 018d899296520a843fc62dd97c87e95f85fbeab9 Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Thu, 24 Sep 2015 07:15:07 -0500 Subject: [PATCH 2/5] Changed how the KeyValueStores use serializers and deserializers --- .../kafka/streams/examples/ProcessorJob.java | 2 +- .../streams/state/InMemoryKeyValueStore.java | 124 +++++++++-- .../streams/state/MeteredKeyValueStore.java | 15 +- .../streams/state/RocksDBKeyValueStore.java | 203 ++++++++++++------ .../apache/kafka/streams/state/Serdes.java | 164 ++++++++++++++ .../internals/ProcessorTopologyTest.java | 2 +- 6 files changed, 421 insertions(+), 89 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/Serdes.java diff --git a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java index 0b3aba8716aed..46c15304c2298 100644 --- a/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java +++ b/streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java @@ -48,7 +48,7 @@ public Processor instance() { public void init(ProcessorContext context) { this.context = context; this.context.schedule(1000); - this.kvStore = new InMemoryKeyValueStore<>("local-state", context); + this.kvStore = InMemoryKeyValueStore.create("local-state", context, String.class, Integer.class); } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java index e9aaa20265a53..8e35d504cb65a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryKeyValueStore.java @@ -18,6 +18,8 @@ package org.apache.kafka.streams.state; import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.utils.SystemTime; import org.apache.kafka.common.utils.Time; @@ -35,26 +37,127 @@ */ public class InMemoryKeyValueStore extends MeteredKeyValueStore { - public InMemoryKeyValueStore(String name, ProcessorContext context) { - this(name, context, new SystemTime()); + /** + * Create an in-memory key value store that records changes to a Kafka topic and the system time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context, Class keyClass, Class valueClass) { + return new InMemoryKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), + new SystemTime()); } - public InMemoryKeyValueStore(String name, ProcessorContext context, Time time) { - super(name, new MemoryStore(name, context), context, "kafka-streams", time); + /** + * Create an in-memory key value store that records changes to a Kafka topic and the given time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context, Class keyClass, Class valueClass, + Time time) { + return new InMemoryKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), + time); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default + * serializers and deserializers, and the system time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context) { + return create(name, context, new SystemTime()); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default + * serializers and deserializers, and the given time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context, Time time) { + return new InMemoryKeyValueStore<>(name, context, new Serdes(name, context), time); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and + * the system time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { + return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime()); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and + * the given time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryKeyValueStore create(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + Time time) { + Serdes serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + return new InMemoryKeyValueStore<>(name, context, serdes, time); + } + + protected InMemoryKeyValueStore(String name, ProcessorContext context, Serdes serdes, Time time) { + super(name, new MemoryStore(name), context, serdes, "kafka-streams", time); } private static class MemoryStore implements KeyValueStore { private final String name; private final NavigableMap map; - private final ProcessorContext context; - @SuppressWarnings("unchecked") - public MemoryStore(String name, ProcessorContext context) { + public MemoryStore(String name) { super(); this.name = name; this.map = new TreeMap<>(); - this.context = context; } @Override @@ -103,11 +206,6 @@ public void flush() { // do-nothing since it is in-memory } - public void restore() { - // this should not happen since it is in-memory, hence no state to load from disk - throw new IllegalStateException("This should not happen"); - } - @Override public void close() { // do-nothing 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 018f1c621c64e..e44c97df708c6 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 @@ -41,6 +41,7 @@ public class MeteredKeyValueStore implements KeyValueStore { protected final KeyValueStore inner; + protected final Serdes serialization; private final Time time; private final String group; @@ -61,8 +62,10 @@ public class MeteredKeyValueStore implements KeyValueStore { private final ProcessorContext context; // always wrap the logged store with the metered store - public MeteredKeyValueStore(final String name, final KeyValueStore inner, ProcessorContext context, String group, Time time) { + public MeteredKeyValueStore(final String name, final KeyValueStore inner, ProcessorContext context, + Serdes serialization, String group, Time time) { this.inner = inner; + this.serialization = serialization; this.time = time; this.group = group; @@ -87,8 +90,8 @@ public MeteredKeyValueStore(final String name, final KeyValueStore inner, // register and possibly restore the state from the logs long startNs = time.nanoseconds(); try { - final Deserializer keyDeserializer = (Deserializer) context.keyDeserializer(); - final Deserializer valDeserializer = (Deserializer) context.valueDeserializer(); + final Deserializer keyDeserializer = serialization.keyDeserializer(); + final Deserializer valDeserializer = serialization.valueDeserializer(); context.register(this, new RestoreFunc() { @Override @@ -216,10 +219,10 @@ public void flush() { private void logChange() { RecordCollector collector = ((ProcessorContextImpl) context).recordCollector(); - Serializer keySerializer = (Serializer) context.keySerializer(); - Serializer valueSerializer = (Serializer) context.valueSerializer(); - if (collector != null) { + Serializer keySerializer = serialization.keySerializer(); + Serializer valueSerializer = serialization.valueSerializer(); + for (K k : this.dirty) { V v = this.inner.get(k); collector.send(new ProducerRecord<>(this.topic, this.partition, k, v), keySerializer, valueSerializer); diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java index aac5a96b6901a..50e3a3614cdea 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java @@ -44,21 +44,123 @@ * @param the type of keys * @param the type of values */ -public class RocksDBKeyValueStore extends MeteredKeyValueStore { +public class RocksDBKeyValueStore extends MeteredKeyValueStore { + + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the system time + * provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context, Class keyClass, Class valueClass) { + return new RocksDBKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), + new SystemTime()); + } + + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage and the given time + * provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param time the time provider; may not be null + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context, Class keyClass, Class valueClass, + Time time) { + return new RocksDBKeyValueStore<>(name, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), + time); + } - public RocksDBKeyValueStore(String name, ProcessorContext context, - Serializer keySerializer, Deserializer keyDeserializer, - Serializer valueSerializer, Deserializer valueDeserializer) { - this(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime()); + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the + * {@link ProcessorContext}'s default serializers and deserializers, and the system time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context) { + return create(name, context, new SystemTime()); } - public RocksDBKeyValueStore(String name, ProcessorContext context, - Serializer keySerializer, Deserializer keyDeserializer, - Serializer valueSerializer, Deserializer valueDeserializer, Time time) { - super(name, new RocksDBStore(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer), context, "kafka-streams", time); + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the + * {@link ProcessorContext}'s default serializers and deserializers, and the given time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param time the time provider; may not be null + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context, Time time) { + return new RocksDBKeyValueStore<>(name, context, new Serdes(name, context), time); } - private static class RocksDBStore implements KeyValueStore { + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the + * supplied serializers and deserializers, and the system time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { + return create(name, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime()); + } + + /** + * Create a key value store that records changes to a Kafka topic and that uses RocksDB for local storage, the + * supplied serializers and deserializers, and the given time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @param time the time provider; may not be null + * @return the key-value store + */ + public static RocksDBKeyValueStore create(String name, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + Time time) { + Serdes serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + return new RocksDBKeyValueStore<>(name, context, serdes, time); + } + + protected RocksDBKeyValueStore(String name, ProcessorContext context, Serdes serdes, Time time) { + super(name, new RocksDBStore(name, context, serdes), context, serdes, "kafka-streams", time); + } + + private static class RocksDBStore implements KeyValueStore { private static final int TTL_NOT_USED = -1; @@ -72,10 +174,7 @@ private static class RocksDBStore implements KeyValueStore { private static final CompactionStyle COMPACTION_STYLE = CompactionStyle.UNIVERSAL; private static final String DB_FILE_DIR = "rocksdb"; - private final Serializer keySerializer; - private final Serializer valueSerializer; - private final Deserializer keyDeserializer; - private final Deserializer valueDeserializer; + private final Serdes serdes; private final String topic; private final int partition; @@ -90,16 +189,11 @@ private static class RocksDBStore implements KeyValueStore { private RocksDB db; - @SuppressWarnings("unchecked") - public RocksDBStore(String name, ProcessorContext context, Serializer keySerializer, Deserializer keyDeserializer, - Serializer valueSerializer, Deserializer valueDeserializer) { + public RocksDBStore(String name, ProcessorContext context, Serdes serdes) { this.topic = name; this.partition = context.id(); this.context = context; - this.keySerializer = keySerializer != null ? keySerializer : (Serializer)context.keySerializer(); - this.keyDeserializer = keyDeserializer != null ? keyDeserializer : (Deserializer)context.keyDeserializer(); - this.valueSerializer = valueSerializer != null ? valueSerializer : (Serializer)context.valueSerializer(); - this.valueDeserializer = valueDeserializer != null ? valueDeserializer : (Deserializer)context.valueDeserializer(); + this.serdes = serdes; // initialize the rocksdb options BlockBasedTableConfig tableConfig = new BlockBasedTableConfig(); @@ -151,24 +245,24 @@ public String name() { public boolean persistent() { return false; } - + @Override public V get(K key) { try { - return valueFrom(this.db.get(rawKey(key))); + return serdes.valueFrom(this.db.get(serdes.rawKey(key))); } catch (RocksDBException e) { // TODO: this needs to be handled more accurately throw new KafkaException("Error while executing get " + key.toString() + " from store " + this.topic, e); } } - + @Override public void put(K key, V value) { try { if (value == null) { - db.remove(wOptions, rawKey(key)); + db.remove(wOptions, serdes.rawKey(key)); } else { - db.put(wOptions, rawKey(key), rawValue(value)); + db.put(wOptions, serdes.rawKey(key), serdes.rawValue(value)); } } catch (RocksDBException e) { // TODO: this needs to be handled more accurately @@ -178,7 +272,7 @@ public void put(K key, V value) { @Override public void putAll(List> entries) { - for (Entry entry : entries) + for (Entry entry : entries) put(entry.key(), entry.value()); } @@ -191,14 +285,14 @@ public V delete(K key) { @Override public KeyValueIterator range(K from, K to) { - return new RocksDBRangeIterator(db.newIterator(), topic, keySerializer , keyDeserializer, valueDeserializer, from, to); + return new RocksDBRangeIterator(db.newIterator(), serdes, from, to); } @Override public KeyValueIterator all() { RocksIterator innerIter = db.newIterator(); innerIter.seekToFirst(); - return new RocksDbIterator(innerIter, topic, keyDeserializer, valueDeserializer); + return new RocksDbIterator(innerIter, serdes); } @Override @@ -217,46 +311,21 @@ public void close() { db.close(); } - protected final V valueFrom( byte[] rawValue ) { - return valueDeserializer.deserialize(topic, rawValue); - } - - protected final byte[] rawKey( K key ) { - return keySerializer.serialize(topic, key); - } - - protected final byte[] rawValue( V value ) { - return valueSerializer.serialize(topic, value); - } - private static class RocksDbIterator implements KeyValueIterator { private final RocksIterator iter; - private final String topic; - private final Deserializer keyDeserializer; - private final Deserializer valueDeserializer; + private final Serdes serdes; - public RocksDbIterator(RocksIterator iter, String topic, - Deserializer keyDeserializer, Deserializer valueDeserializer) { + public RocksDbIterator(RocksIterator iter, Serdes serdes) { this.iter = iter; - this.topic = topic; - this.keyDeserializer = keyDeserializer; - this.valueDeserializer = valueDeserializer; + this.serdes = serdes; } - protected byte[] peekKey() { + protected byte[] peekRawKey() { return iter.key(); } - - protected final K keyFrom( byte[] rawKey ) { - return keyDeserializer.deserialize(topic, rawKey); - } - - protected final V valueFrom( byte[] rawValue ) { - return valueDeserializer.deserialize(topic, rawValue); - } protected Entry getEntry() { - return new Entry<>(keyFrom(iter.key()),valueFrom(iter.value())); + return new Entry<>(serdes.keyFrom(iter.key()), serdes.valueFrom(iter.value())); } @Override @@ -300,25 +369,23 @@ public int compare(byte[] left, byte[] right) { } } - private static class RocksDBRangeIterator extends RocksDbIterator { + private static class RocksDBRangeIterator extends RocksDbIterator { // RocksDB's JNI interface does not expose getters/setters that allow the // comparator to be pluggable, and the default is lexicographic, so it's // safe to just force lexicographic comparator here for now. private final Comparator comparator = new LexicographicComparator(); byte[] to; - public RocksDBRangeIterator(RocksIterator iter, String topic, - Serializer keySerializer, Deserializer keyDeserializer, - Deserializer valueDeserializer, - K from, K to) { - super(iter,topic,keyDeserializer,valueDeserializer); - iter.seek(keySerializer.serialize(topic, from)); - this.to = keySerializer.serialize(topic, to); + public RocksDBRangeIterator(RocksIterator iter, Serdes serdes, + K from, K to) { + super(iter, serdes); + iter.seek(serdes.rawKey(from)); + this.to = serdes.rawKey(to); } @Override public boolean hasNext() { - return super.hasNext() && comparator.compare(super.peekKey(), this.to) < 0; + return super.hasNext() && comparator.compare(super.peekRawKey(), this.to) < 0; } } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java new file mode 100644 index 0000000000000..1ebfe88744864 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java @@ -0,0 +1,164 @@ +/** + * 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.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.LongSerializer; +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.processor.ProcessorContext; + +final class Serdes { + + public static Serdes withBuiltinTypes(String topic, Class keyClass, Class valueClass) { + Serializer keySerializer = serializer(keyClass); + Deserializer keyDeserializer = deserializer(keyClass); + Serializer valueSerializer = serializer(valueClass); + Deserializer valueDeserializer = deserializer(valueClass); + return new Serdes<>(topic, keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + } + + @SuppressWarnings("unchecked") + private static Serializer serializer(Class type) { + if (String.class.isAssignableFrom(type)) return (Serializer) new StringSerializer(); + if (Integer.class.isAssignableFrom(type)) return (Serializer) new IntegerSerializer(); + if (Long.class.isAssignableFrom(type)) return (Serializer) new LongSerializer(); + if (byte[].class.isAssignableFrom(type)) return (Serializer) new ByteArraySerializer(); + throw new IllegalArgumentException("Unknown class for built-in serializer"); + } + + @SuppressWarnings("unchecked") + private static Deserializer deserializer(Class type) { + if (String.class.isAssignableFrom(type)) return (Deserializer) new StringDeserializer(); + if (Integer.class.isAssignableFrom(type)) return (Deserializer) new IntegerDeserializer(); + if (Long.class.isAssignableFrom(type)) return (Deserializer) new LongDeserializer(); + if (byte[].class.isAssignableFrom(type)) return (Deserializer) new ByteArrayDeserializer(); + throw new IllegalArgumentException("Unknown class for built-in serializer"); + } + + private final String topic; + private final Serializer keySerializer; + private final Serializer valueSerializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; + + /** + * Create a context for serialization using the specified serializers and deserializers. + * + * @param topic the name of the topic + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + */ + public Serdes(String topic, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { + this.topic = topic; + this.keySerializer = keySerializer; + this.keyDeserializer = keyDeserializer; + this.valueSerializer = valueSerializer; + this.valueDeserializer = valueDeserializer; + } + + /** + * Create a context for serialization using the specified serializers and deserializers, or if any of them are null the + * corresponding {@link ProcessorContext}'s default serializer or deserializer, which + * must match the key and value types used as parameters for this object. + * + * @param topic the name of the topic + * @param keySerializer the serializer for keys; may be null if the {@link ProcessorContext#keySerializer() default + * key serializer} should be used + * @param keyDeserializer the deserializer for keys; may be null if the {@link ProcessorContext#keyDeserializer() default + * key deserializer} should be used + * @param valueSerializer the serializer for values; may be null if the {@link ProcessorContext#valueSerializer() default + * value serializer} should be used + * @param valueDeserializer the deserializer for values; may be null if the {@link ProcessorContext#valueDeserializer() + * default value deserializer} should be used + * @param context the processing context + */ + @SuppressWarnings("unchecked") + public Serdes(String topic, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + ProcessorContext context) { + this.topic = topic; + this.keySerializer = keySerializer != null ? keySerializer : (Serializer) context.keySerializer(); + this.keyDeserializer = keyDeserializer != null ? keyDeserializer : (Deserializer) context.keyDeserializer(); + this.valueSerializer = valueSerializer != null ? valueSerializer : (Serializer) context.valueSerializer(); + this.valueDeserializer = valueDeserializer != null ? valueDeserializer : (Deserializer) context.valueDeserializer(); + } + + /** + * Create a context for serialization using the {@link ProcessorContext}'s default serializers and deserializers, which + * must match the key and value types used as parameters for this object. + * + * @param topic the name of the topic + * @param context the processing context + */ + @SuppressWarnings("unchecked") + public Serdes(String topic, + ProcessorContext context) { + this.topic = topic; + this.keySerializer = (Serializer) context.keySerializer(); + this.keyDeserializer = (Deserializer) context.keyDeserializer(); + this.valueSerializer = (Serializer) context.valueSerializer(); + this.valueDeserializer = (Deserializer) context.valueDeserializer(); + } + + public Deserializer keyDeserializer() { + return keyDeserializer; + } + + public Serializer keySerializer() { + return keySerializer; + } + + public Deserializer valueDeserializer() { + return valueDeserializer; + } + + public Serializer valueSerializer() { + return valueSerializer; + } + + public String topic() { + return topic; + } + + public K keyFrom(byte[] rawKey) { + return keyDeserializer.deserialize(topic, rawKey); + } + + public V valueFrom(byte[] rawValue) { + return valueDeserializer.deserialize(topic, rawValue); + } + + public byte[] rawKey(K key) { + return keySerializer.serialize(topic, key); + } + + public byte[] rawValue(V value) { + return valueSerializer.serialize(topic, value); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java index 1abb989e30de2..af66ebf9fcd61 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java @@ -289,7 +289,7 @@ public StatefulProcessor(String storeName) { @Override public void init(ProcessorContext context) { super.init(context); - store = new InMemoryKeyValueStore<>(storeName, context); + store = InMemoryKeyValueStore.create(storeName, context, String.class, String.class); } @Override From 07bca6aa2dd5b0be52606e9bc6a5a968145ab428 Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Thu, 24 Sep 2015 15:48:46 -0500 Subject: [PATCH 3/5] Corrected RocksDBKeyValueStore initialiation logic to create the parent directory if missing --- .../org/apache/kafka/streams/state/RocksDBKeyValueStore.java | 1 + 1 file changed, 1 insertion(+) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java index 50e3a3614cdea..079c223cce575 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/RocksDBKeyValueStore.java @@ -224,6 +224,7 @@ public RocksDBStore(String name, ProcessorContext context, Serdes serdes) private RocksDB openDB(File dir, Options options, int ttl) { try { if (ttl == TTL_NOT_USED) { + dir.getParentFile().mkdirs(); return RocksDB.open(options, dir.toString()); } else { throw new KafkaException("Change log is not supported for store " + this.topic + " since it is TTL based."); From 19680dd10e27485610a48a5a70ddabe4fdd98bcd Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Thu, 24 Sep 2015 16:00:26 -0500 Subject: [PATCH 4/5] Removed MeteredKeyValueStore's dependency on ProcessorContextImpl via new RecordCollectorSupplier interface --- .../internals/ProcessorContextImpl.java | 3 +- .../processor/internals/RecordCollector.java | 4 ++ .../streams/state/MeteredKeyValueStore.java | 2 +- .../kafka/test/MockProcessorContext.java | 52 +++++++++++++++---- 4 files changed, 50 insertions(+), 11 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java index b3502220cf3ed..8541acbc27a68 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java @@ -37,7 +37,7 @@ import java.util.Map; import java.util.Set; -public class ProcessorContextImpl implements ProcessorContext { +public class ProcessorContextImpl implements ProcessorContext, RecordCollector.Supplier { private static final Logger log = LoggerFactory.getLogger(ProcessorContextImpl.class); @@ -75,6 +75,7 @@ public ProcessorContextImpl(int id, this.initialized = false; } + @Override public RecordCollector recordCollector() { return this.collector; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java index ad2f647a2ac3d..5d56a5bf520fd 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java @@ -30,6 +30,10 @@ import java.util.Map; public class RecordCollector { + + public static interface Supplier { + public RecordCollector recordCollector(); + } private static final Logger log = LoggerFactory.getLogger(RecordCollector.class); 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 e44c97df708c6..ae5a1fa7d29d0 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 @@ -218,7 +218,7 @@ public void flush() { } private void logChange() { - RecordCollector collector = ((ProcessorContextImpl) context).recordCollector(); + RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector(); if (collector != null) { Serializer keySerializer = serialization.keySerializer(); Serializer valueSerializer = serialization.valueSerializer(); diff --git a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java index 3fdfc82c1ff76..77e9e3aa623e0 100644 --- a/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java +++ b/streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java @@ -23,31 +23,65 @@ import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.processor.internals.RecordCollector; import java.io.File; import java.util.HashMap; import java.util.Map; -public class MockProcessorContext implements ProcessorContext { +public class MockProcessorContext implements ProcessorContext, RecordCollector.Supplier { private final KStreamTestDriver driver; - private final Serializer serializer; - private final Deserializer deserializer; + private final Serializer keySerializer; + private final Serializer valueSerializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; + private final RecordCollector.Supplier recordCollectorSupplier; private Map storeMap = new HashMap<>(); long timestamp = -1L; public MockProcessorContext(KStreamTestDriver driver, Serializer serializer, Deserializer deserializer) { + this(driver, serializer, deserializer, serializer, deserializer, (RecordCollector.Supplier) null); + } + + public MockProcessorContext(KStreamTestDriver driver, Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + final RecordCollector collector) { + this(driver, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, + collector == null ? null : new RecordCollector.Supplier() { + @Override + public RecordCollector recordCollector() { + return collector; + } + }); + } + + public MockProcessorContext(KStreamTestDriver driver, Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + RecordCollector.Supplier collectorSupplier) { this.driver = driver; - this.serializer = serializer; - this.deserializer = deserializer; + this.keySerializer = keySerializer; + this.valueSerializer = valueSerializer; + this.keyDeserializer = keyDeserializer; + this.valueDeserializer = valueDeserializer; + this.recordCollectorSupplier = collectorSupplier; + } + + @Override + public RecordCollector recordCollector() { + if (recordCollectorSupplier == null) { + throw new UnsupportedOperationException("No RecordCollector specified"); + } + return recordCollectorSupplier.recordCollector(); } public void setTime(long timestamp) { this.timestamp = timestamp; } + @Override public int id() { return -1; } @@ -59,22 +93,22 @@ public boolean joinable() { @Override public Serializer keySerializer() { - return serializer; + return keySerializer; } @Override public Serializer valueSerializer() { - return serializer; + return valueSerializer; } @Override public Deserializer keyDeserializer() { - return deserializer; + return keyDeserializer; } @Override public Deserializer valueDeserializer() { - return deserializer; + return valueDeserializer; } @Override From 50606fe6b332ab129bf5666e68a5eb8aa71701df Mon Sep 17 00:00:00 2001 From: Randall Hauch Date: Thu, 24 Sep 2015 16:02:53 -0500 Subject: [PATCH 5/5] Added InMemoryLRUCacheStore and changed MeteredKeyValueStore to support removes from the inner store. Also added test framework and unit tests for existing KeyValueStore implementations. --- .../streams/state/InMemoryLRUCacheStore.java | 294 +++++++++++++ .../streams/state/MeteredKeyValueStore.java | 59 ++- .../state/AbstractKeyValueStoreTest.java | 223 ++++++++++ .../state/InMemoryKeyValueStoreTest.java | 31 ++ .../state/InMemoryLRUCacheStoreTest.java | 132 ++++++ .../state/KeyValueStoreTestDriver.java | 407 ++++++++++++++++++ .../state/RocksDBKeyValueStoreTest.java | 31 ++ 7 files changed, 1163 insertions(+), 14 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/AbstractKeyValueStoreTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.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..794ff80cc1d4e --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/InMemoryLRUCacheStore.java @@ -0,0 +1,294 @@ +/** + * 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.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; +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 { + + /** + * Create an in-memory key value store that records changes to a Kafka topic and the system time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be a power of 2 + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, + ProcessorContext context, Class keyClass, Class valueClass) { + return create(name, capacity, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), new SystemTime()); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic and the given time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be one less than a power of 2 + * @param context the processing context + * @param keyClass the class for the keys, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param valueClass the class for the values, which must be one of the types for which Kafka has built-in serializers and + * deserializers (e.g., {@code String.class}, {@code Integer.class}, {@code Long.class}, or + * {@code byte[].class}) + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, + ProcessorContext context, Class keyClass, Class valueClass, + Time time) { + return create(name, capacity, context, Serdes.withBuiltinTypes(name, keyClass, valueClass), time); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default + * serializers and deserializers, and the system time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be one less than a power of 2 + * @param context the processing context + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, ProcessorContext context) { + return create(name, capacity, context, new SystemTime()); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the {@link ProcessorContext}'s default + * serializers and deserializers, and the given time provider. + *

+ * NOTE: the default serializers and deserializers in the context must match the key and value types + * used as parameters for this key value store. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be one less than a power of 2 + * @param context the processing context + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, ProcessorContext context, Time time) { + return create(name, capacity, context, new Serdes(name, context), time); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and + * the system time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be one less than a power of 2 + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer) { + return create(name, capacity, context, keySerializer, keyDeserializer, valueSerializer, valueDeserializer, new SystemTime()); + } + + /** + * Create an in-memory key value store that records changes to a Kafka topic, the provided serializers and deserializers, and + * the given time provider. + * + * @param name the name of the store, used in the name of the topic to which entries are recorded + * @param capacity the maximum capacity of the LRU cache; should be one less than a power of 2 + * @param context the processing context + * @param keySerializer the serializer for keys; may not be null + * @param keyDeserializer the deserializer for keys; may not be null + * @param valueSerializer the serializer for values; may not be null + * @param valueDeserializer the deserializer for values; may not be null + * @param time the time provider; may not be null + * @return the key-value store + */ + public static InMemoryLRUCacheStore create(String name, int capacity, ProcessorContext context, + Serializer keySerializer, Deserializer keyDeserializer, + Serializer valueSerializer, Deserializer valueDeserializer, + Time time) { + Serdes serdes = new Serdes<>(name, keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + return create(name, capacity, context, serdes, time); + } + + + protected static InMemoryLRUCacheStore create (String name, int capacity, ProcessorContext context, + Serdes serdes, Time time) { + 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; + + } + public 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 + } + } + } +} 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 ae5a1fa7d29d0..72bf5961a0e65 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 @@ -58,12 +58,14 @@ 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 public MeteredKeyValueStore(final String name, final KeyValueStore inner, ProcessorContext context, - Serdes serialization, String group, Time time) { + Serdes serialization, String group, Time time) { this.inner = inner; this.serialization = serialization; @@ -85,7 +87,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(); @@ -97,7 +101,7 @@ public MeteredKeyValueStore(final String name, final KeyValueStore inner, @Override public void apply(byte[] key, byte[] value) { inner.put(keyDeserializer.deserialize(topic, key), - valDeserializer.deserialize(topic, value)); + valDeserializer.deserialize(topic, value)); } }); } finally { @@ -114,9 +118,14 @@ private Sensor createSensor(String storeName, String operation) { } private void addLatencyMetrics(Sensor sensor, String opName, String... kvs) { - maybeAddMetric(sensor, new MetricName(opName + "-avg-latency-ms", group, "The average latency in milliseconds of the key-value store operation.", kvs), new Avg()); - maybeAddMetric(sensor, new MetricName(opName + "-max-latency-ms", group, "The max latency in milliseconds of the key-value store operation.", kvs), new Max()); - maybeAddMetric(sensor, new MetricName(opName + "-qps", group, "The average number of occurance of the given key-value store operation per second.", kvs), new Rate(new Count())); + maybeAddMetric(sensor, new MetricName(opName + "-avg-latency-ms", group, + "The average latency in milliseconds of the key-value store operation.", kvs), new Avg()); + maybeAddMetric(sensor, new MetricName(opName + "-max-latency-ms", group, + "The max latency in milliseconds of the key-value store operation.", kvs), new Max()); + maybeAddMetric(sensor, + new MetricName(opName + "-qps", group, + "The average number of occurance of the given key-value store operation per second.", kvs), + new Rate(new Count())); } private void maybeAddMetric(Sensor sensor, MetricName name, MeasurableStat stat) { @@ -151,8 +160,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 { recordLatency(this.putTime, startNs, time.nanoseconds()); } @@ -165,11 +174,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 { recordLatency(this.putAllTime, startNs, time.nanoseconds()); } @@ -181,9 +191,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 { @@ -191,6 +201,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); @@ -217,16 +239,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/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java new file mode 100644 index 0000000000000..518dd34a7f4b0 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/AbstractKeyValueStoreTest.java @@ -0,0 +1,223 @@ +/** + * 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 static org.junit.Assert.fail; + +import org.apache.kafka.streams.processor.ProcessorContext; +import org.junit.After; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; + +public abstract class AbstractKeyValueStoreTest { + + protected static final File STATE_DIR = new File("build/data").getAbsoluteFile(); + + protected abstract KeyValueStore createKeyValueStore(ProcessorContext context, + Class keyClass, Class valueClass, + boolean useContextSerdes); + + @After + public void cleanup() { + if (STATE_DIR.exists()) { + try { + Files.walkFileTree(STATE_DIR.toPath(), new SimpleFileVisitor() { + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { + Files.delete(file); + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException { + Files.delete(dir); + return FileVisitResult.CONTINUE; + } + + }); + } catch (IOException e) { + // do nothing + } + } + } + + @Test + public void testIntegerKeysAndStringValues() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(); + driver.useStateDir(STATE_DIR); + KeyValueStore store = createKeyValueStore(driver.context(), Integer.class, String.class, false); + try { + + // Verify that the store reads and writes correctly ... + store.put(0, "zero"); + store.put(1, "one"); + store.put(2, "two"); + store.put(4, "four"); + store.put(5, "five"); + assertEquals(5, driver.sizeOf(store)); + assertEquals("zero", store.get(0)); + assertEquals("one", store.get(1)); + assertEquals("two", store.get(2)); + assertNull(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(); + assertEquals("zero", driver.flushedEntryStored(0)); + assertEquals("one", driver.flushedEntryStored(1)); + assertEquals("two", driver.flushedEntryStored(2)); + assertEquals("four", driver.flushedEntryStored(4)); + assertEquals(null, driver.flushedEntryStored(5)); + + assertEquals(false, driver.flushedEntryRemoved(0)); + assertEquals(false, driver.flushedEntryRemoved(1)); + assertEquals(false, driver.flushedEntryRemoved(2)); + assertEquals(false, driver.flushedEntryRemoved(4)); + assertEquals(true, driver.flushedEntryRemoved(5)); + + // Check range iteration ... + try ( KeyValueIterator iter = store.range(2, 4)) { + while ( iter.hasNext() ) { + Entry entry = iter.next(); + if ( entry.key().equals(2) ) assertEquals("two", entry.value()); + else if ( entry.key().equals(4) ) assertEquals("four", entry.value()); + else fail("Unexpected entry: " + entry); + } + } + + // Check range iteration ... + try ( KeyValueIterator iter = store.range(2, 6)) { + while ( iter.hasNext() ) { + Entry entry = iter.next(); + if ( entry.key().equals(2) ) assertEquals("two", entry.value()); + else if ( entry.key().equals(4) ) assertEquals("four", entry.value()); + else fail("Unexpected entry: " + entry); + } + } + } finally { + store.close(); + } + } + + @Test + public void testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializers() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(Integer.class, String.class); + driver.useStateDir(STATE_DIR); + KeyValueStore store = createKeyValueStore(driver.context(), Integer.class, String.class, true); + try { + + // Verify that the store reads and writes correctly ... + store.put(0, "zero"); + store.put(1, "one"); + store.put(2, "two"); + store.put(4, "four"); + store.put(5, "five"); + assertEquals(5, driver.sizeOf(store)); + assertEquals("zero", store.get(0)); + assertEquals("one", store.get(1)); + assertEquals("two", store.get(2)); + assertNull(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(); + assertEquals("zero", driver.flushedEntryStored(0)); + assertEquals("one", driver.flushedEntryStored(1)); + assertEquals("two", driver.flushedEntryStored(2)); + assertEquals("four", driver.flushedEntryStored(4)); + assertEquals(null, driver.flushedEntryStored(5)); + + assertEquals(false, driver.flushedEntryRemoved(0)); + assertEquals(false, driver.flushedEntryRemoved(1)); + assertEquals(false, driver.flushedEntryRemoved(2)); + assertEquals(false, driver.flushedEntryRemoved(4)); + assertEquals(true, driver.flushedEntryRemoved(5)); + } finally { + store.close(); + } + } + + @Test + public void testRestoringInetgerKeysAndValues() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(Integer.class, String.class); + driver.useStateDir(STATE_DIR); + + // Add any entries that will be restored to any store + // that uses the driver's context ... + driver.addRestoreEntry(0, "zero"); + driver.addRestoreEntry(1, "one"); + driver.addRestoreEntry(2, "two"); + driver.addRestoreEntry(4, "four"); + + // Create the store, which should register with the context and automatically + // receive the restore entries ... + KeyValueStore store = createKeyValueStore(driver.context(), Integer.class, String.class, false); + try { + // Verify that the store's contents were properly restored ... + assertEquals(0, driver.checkForRestoredEntries(store)); + + // and there are no other entries ... + assertEquals(4, driver.sizeOf(store)); + } finally { + store.close(); + } + } + + @Test + public void testRestoringInetgerKeysAndValuesUsingDefaultSerializersAndDeserializers() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(Integer.class, String.class); + driver.useStateDir(STATE_DIR); + + // Add any entries that will be restored to any store + // that uses the driver's context ... + driver.addRestoreEntry(0, "zero"); + driver.addRestoreEntry(1, "one"); + driver.addRestoreEntry(2, "two"); + driver.addRestoreEntry(4, "four"); + + // Create the store, which should register with the context and automatically + // receive the restore entries ... + KeyValueStore store = createKeyValueStore(driver.context(), Integer.class, String.class, true); + try { + // Verify that the store's contents were properly restored ... + assertEquals(0, driver.checkForRestoredEntries(store)); + + // and there are no other entries ... + assertEquals(4, driver.sizeOf(store)); + } finally { + store.close(); + } + } + +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java new file mode 100644 index 0000000000000..aff6091cefcb5 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryKeyValueStoreTest.java @@ -0,0 +1,31 @@ +/** + * 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.streams.processor.ProcessorContext; + +public class InMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest { + + @Override + protected KeyValueStore createKeyValueStore(ProcessorContext context, Class keyClass, Class valueClass, + boolean useContextSerdes) { + if ( useContextSerdes ) { + return InMemoryKeyValueStore.create("my-store", context); + } + return InMemoryKeyValueStore.create("my-store", context, keyClass, valueClass); + } +} 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..856381905911b --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/InMemoryLRUCacheStoreTest.java @@ -0,0 +1,132 @@ +/** + * 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.junit.Test; + +public class InMemoryLRUCacheStoreTest { + + @Test + public void testIntegerKeysAndStringValues() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(); + InMemoryLRUCacheStore store = InMemoryLRUCacheStore.create("my-store", 3, driver.context(), + Integer.class, String.class); + + // 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 testIntegerKeysAndStringValuesUsingDefaultSerializersAndDeserializers() { + // Create the test driver ... + KeyValueStoreTestDriver driver = KeyValueStoreTestDriver.create(Integer.class, String.class); + InMemoryLRUCacheStore store = InMemoryLRUCacheStore.create("my-store", 3, driver.context()); + + // 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 testRestoringInetgerKeysAndValues() { + // 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.addRestoreEntry(1, "one"); + driver.addRestoreEntry(2, "two"); + driver.addRestoreEntry(4, "four"); + + // Create the store, which should register with the context and automatically + // receive the restore entries ... + InMemoryLRUCacheStore store = InMemoryLRUCacheStore.create("my-store", 3, driver.context(), + Integer.class, String.class); + + // 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)); + } + +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java new file mode 100644 index 0000000000000..8bdd702057417 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java @@ -0,0 +1,407 @@ +/** + * 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.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.RestoreFunc; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.internals.RecordCollector; +import org.apache.kafka.test.MockProcessorContext; + +import java.io.File; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +/** + * A component that provides a {@link #context() ProcessingContext} that can be supplied to a {@link KeyValueStore} so that + * all entries written to the Kafka topic by the store during {@link KeyValueStore#flush()} are captured for testing purposes. + * This class simplifies testing of various {@link KeyValueStore} instances, especially those that use + * {@link MeteredKeyValueStore} to monitor and write its entries to the Kafka topic. + *

+ *

Basic usage

+ * This component can be used to help test a {@link KeyValueStore}'s ability to read and write entries. + * + * 
+ * // Create the test driver ...
+ * KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create();
+ * InMemoryKeyValueStore<Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
+ *                                                                             Integer.class, String.class);
+ * 
+ * // Verify that the store reads and writes correctly ...
+ * store.put(0, "zero");
+ * store.put(1, "one");
+ * store.put(2, "two");
+ * store.put(4, "four");
+ * store.put(5, "five");
+ * assertEquals(5, driver.sizeOf(store));
+ * assertEquals("zero", store.get(0));
+ * assertEquals("one", store.get(1));
+ * assertEquals("two", store.get(2));
+ * assertEquals("four", store.get(4));
+ * assertEquals("five", store.get(5));
+ * assertNull(store.get(3));
+ * store.delete(5);
+ * 
+ * // Flush the store and verify all current entries were properly flushed ...
+ * store.flush();
+ * assertEquals("zero", driver.flushedEntryStored(0));
+ * assertEquals("one", driver.flushedEntryStored(1));
+ * assertEquals("two", driver.flushedEntryStored(2));
+ * assertEquals("four", driver.flushedEntryStored(4));
+ * assertEquals(null, driver.flushedEntryStored(5));
+ * 
+ * assertEquals(false, driver.flushedEntryRemoved(0));
+ * assertEquals(false, driver.flushedEntryRemoved(1));
+ * assertEquals(false, driver.flushedEntryRemoved(2));
+ * assertEquals(false, driver.flushedEntryRemoved(4));
+ * assertEquals(true, driver.flushedEntryRemoved(5));
+ *
+ * + *

+ *

Restoring a store

+ * This component can be used to test whether a {@link KeyValueStore} implementation properly + * {@link ProcessorContext#register(StateStore, RestoreFunc) registers itself} with the {@link ProcessorContext}. To do this, + * create an instance of this driver component, {@link #addRestoreEntry(Object, Object) add entries} that will be passed to the + * store upon creation, and then create the store using this driver's {@link #context() ProcessorContext}: + * + * 
+ * // Create the test driver ...
+ * KeyValueStoreTestDriver<Integer, String> driver = KeyValueStoreTestDriver.create(Integer.class, String.class);
+ * 
+ * // Add any entries that will be restored to any store
+ * // that uses the driver's context ...
+ * driver.addRestoreEntry(0, "zero");
+ * driver.addRestoreEntry(1, "one");
+ * driver.addRestoreEntry(2, "two");
+ * driver.addRestoreEntry(4, "four");
+ * 
+ * // Create the store, which should register with the context and automatically
+ * // receive the restore entries ...
+ * InMemoryKeyValueStore<Integer, String> store = InMemoryKeyValueStore.create("my-store", driver.context(),
+ *                                                                             Integer.class, String.class);
+ * 
+ * // Verify that the store's contents were properly restored ...
+ * assertEquals(0, driver.checkForRestoredEntries(store));
+ * 
+ * // and there are no other entries ...
+ * assertEquals(4, driver.sizeOf(store));
+ *
+ * + * @param the type of keys placed in the store + * @param the type of values placed in the store + */ +public class KeyValueStoreTestDriver { + + private static Serializer unusableSerializer() { + return new Serializer() { + @Override + public void configure(Map configs, boolean isKey) { + } + + @Override + public byte[] serialize(String topic, T data) { + throw new UnsupportedOperationException("This serializer should not be used"); + } + + @Override + public void close() { + } + }; + }; + + private static Deserializer unusableDeserializer() { + return new Deserializer() { + @Override + public void configure(Map configs, boolean isKey) { + } + + @Override + public T deserialize(String topic, byte[] data) { + throw new UnsupportedOperationException("This serializer should not be used"); + } + + @Override + public void close() { + } + }; + }; + + /** + * Create a driver object that will have a {@link #context()} that records messages + * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides unusable default key and + * value serializers and deserializers. This can be used when the actual serializers and deserializers are supplied to the + * store during creation, which should eliminate the need for a store to depend on the ProcessorContext's default key and + * value serializers and deserializers. + * + * @return the test driver; never null + */ + public static KeyValueStoreTestDriver create() { + Serializer keySerializer = unusableSerializer(); + Deserializer keyDeserializer = unusableDeserializer(); + Serializer valueSerializer = unusableSerializer(); + Deserializer valueDeserializer = unusableDeserializer(); + Serdes serdes = new Serdes("unexpected", keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + return new KeyValueStoreTestDriver(serdes); + } + + /** + * Create a driver object that will have a {@link #context()} that records messages + * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides default serializers and + * deserializers for the given built-in key and value types (e.g., {@code String.class}, {@code Integer.class}, + * {@code Long.class}, and {@code byte[].class}). This can be used when store is created to rely upon the + * ProcessorContext's default key and value serializers and deserializers. + * + * @param keyClass the class for the keys; must be one of {@code String.class}, {@code Integer.class}, + * {@code Long.class}, or {@code byte[].class} + * @param valueClass the class for the values; must be one of {@code String.class}, {@code Integer.class}, + * {@code Long.class}, or {@code byte[].class} + * @return the test driver; never null + */ + public static KeyValueStoreTestDriver create(Class keyClass, Class valueClass) { + Serdes serdes = Serdes.withBuiltinTypes("unexpected", keyClass, valueClass); + return new KeyValueStoreTestDriver(serdes); + } + + /** + * Create a driver object that will have a {@link #context()} that records messages + * {@link ProcessorContext#forward(Object, Object) forwarded} by the store and that provides the specified serializers and + * deserializers. This can be used when store is created to rely upon the ProcessorContext's default key and value serializers + * and deserializers. + * + * @param keySerializer the key serializer for the {@link ProcessorContext}; may not be null + * @param keyDeserializer the key deserializer for the {@link ProcessorContext}; may not be null + * @param valueSerializer the value serializer for the {@link ProcessorContext}; may not be null + * @param valueDeserializer the value deserializer for the {@link ProcessorContext}; may not be null + * @return the test driver; never null + */ + public static KeyValueStoreTestDriver create(Serializer keySerializer, + Deserializer keyDeserializer, + Serializer valueSerializer, + Deserializer valueDeserializer) { + Serdes serdes = new Serdes("unexpected", keySerializer, keyDeserializer, valueSerializer, valueDeserializer); + return new KeyValueStoreTestDriver(serdes); + } + + private final Serdes serdes; + private final Map flushedEntries = new HashMap<>(); + private final Set flushedRemovals = new HashSet<>(); + private final List> restorableEntries = new LinkedList<>(); + private final MockProcessorContext context; + private final Map storeMap = new HashMap<>(); + private final Metrics metrics = new Metrics(); + private final RecordCollector recordCollector; + private File stateDir = new File("build/data").getAbsoluteFile(); + + protected KeyValueStoreTestDriver(Serdes serdes) { + this.serdes = serdes; + ByteArraySerializer rawSerializer = new ByteArraySerializer(); + Producer producer = new MockProducer(true, rawSerializer, rawSerializer); + this.recordCollector = new RecordCollector(producer) { + @Override + public void send(ProducerRecord record, Serializer keySerializer, Serializer valueSerializer) { + recordFlushed(record.key(), record.value()); + } + }; + this.context = new MockProcessorContext(null, serdes.keySerializer(), serdes.keyDeserializer(), serdes.valueSerializer(), + serdes.valueDeserializer(), recordCollector) { + @Override + public int id() { + return 1; + } + + @Override + public void forward(K1 key, V1 value, int childIndex) { + forward(key, value); + } + + @Override + public void register(StateStore store, RestoreFunc func) { + storeMap.put(store.name(), store); + restoreEntries(func); + } + + @Override + public StateStore getStateStore(String name) { + return storeMap.get(name); + } + + @Override + public Metrics metrics() { + return metrics; + } + + @Override + public File stateDir() { + if (stateDir == null) { + throw new UnsupportedOperationException("No state directory set"); + } + stateDir.mkdirs(); + return stateDir; + } + }; + } + + /** + * Set the directory that should be used by the store for local disk storage. + * + * @param dir the directory; may be null if no local storage is allowed + */ + public void useStateDir(File dir) { + this.stateDir = dir; + } + + @SuppressWarnings("unchecked") + protected void recordFlushed(K1 key, V1 value) { + K k = (K) key; + if (value == null) { + // This is a removal ... + flushedRemovals.add(k); + flushedEntries.remove(k); + } else { + // This is a normal add + flushedEntries.put(k, (V) value); + flushedRemovals.remove(k); + } + } + + private void restoreEntries(RestoreFunc func) { + for (Entry entry : restorableEntries) { + if (entry != null) { + byte[] rawKey = serdes.rawKey(entry.key()); + byte[] rawValue = serdes.rawValue(entry.value()); + func.apply(rawKey, rawValue); + } + } + } + + /** + * This method adds an entry to the "restore log" for the {@link KeyValueStore}. This should be called before + * creating + * the {@link KeyValueStore} with the {@link #context() ProcessorContext}; when the {@link KeyValueStore} is created, it will + * {@link ProcessorContext#register(StateStore, RestoreFunc) register} itself with the {@link #context() ProcessorContext}, + * and this object will then pre-populate the store with all restore entries added via this method. + * + * @param key the key for the entry + * @param value the value for the entry + */ + public void addRestoreEntry(K key, V value) { + restorableEntries.add(new Entry(key, value)); + } + + /** + * Get the context that should be supplied to a {@link KeyValueStore}'s constructor. This context records any messages + * written by the store to the Kafka topic, making them available via the {@link #flushedEntryStored(Object)} and + * {@link #flushedEntryRemoved(Object)} methods. + *

+ * If the {@link KeyValueStore}'s are to be restored upon its startup, be sure to {@link #addRestoreEntry(Object, Object) + * add the restore entries} before creating the store with the {@link ProcessorContext} returned by this method. + * + * @return the processing context; never null + * @see #addRestoreEntry(Object, Object) + */ + public ProcessorContext context() { + return context; + } + + /** + * Get the entries that are restored to a KeyValueStore when it is constructed with this driver's {@link #context() + * ProcessorContext}. + * + * @return the restore entries; never null but possibly a null iterator + */ + public Iterable> restoredEntries() { + return restorableEntries; + } + + /** + * Utility method that will count the number of {@link #addRestoreEntry(Object, Object) restore entries} missing from the + * supplied store. + * + * @param store the store that is to have all of the {@link #restoredEntries() restore entries} + * @return the number of restore entries missing from the store, or 0 if all restore entries were found + */ + public int checkForRestoredEntries(KeyValueStore store) { + int missing = 0; + for (Entry entry : restorableEntries) { + if (entry != null) { + V value = store.get(entry.key()); + if (!Objects.equals(value, entry.value())) { + ++missing; + } + } + } + return missing; + } + + /** + * Utility method to compute the number of entries within the store. + * + * @param store the key value store using this {@link #context()}. + * @return the number of entries + */ + public int sizeOf(KeyValueStore store) { + int size = 0; + for (KeyValueIterator iterator = store.all(); iterator.hasNext();) { + iterator.next(); + ++size; + } + return size; + } + + /** + * Retrieve the value that the store {@link KeyValueStore#flush() flushed} with the given key. + * + * @param key the key + * @return the value that was flushed with the key, or {@code null} if no such key was flushed or if the entry with this + * key was {@link #flushedEntryStored(Object) removed} upon flush + */ + public V flushedEntryStored(K key) { + return flushedEntries.get(key); + } + + /** + * Determine whether the store {@link KeyValueStore#flush() flushed} the removal of the given key. + * + * @param key the key + * @return {@code true} if the entry with the given key was removed when flushed, or {@code false} if the entry was not + * removed when last flushed + */ + public boolean flushedEntryRemoved(K key) { + return flushedRemovals.contains(key); + } + + /** + * Remove all {@link #flushedEntryStored(Object) flushed entries}, {@link #flushedEntryRemoved(Object) flushed removals}, + */ + public void clear() { + restorableEntries.clear(); + flushedEntries.clear(); + flushedRemovals.clear(); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java new file mode 100644 index 0000000000000..f911990c30aa3 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/state/RocksDBKeyValueStoreTest.java @@ -0,0 +1,31 @@ +/** + * 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.streams.processor.ProcessorContext; + +public class RocksDBKeyValueStoreTest extends AbstractKeyValueStoreTest { + + @Override + protected KeyValueStore createKeyValueStore(ProcessorContext context, Class keyClass, Class valueClass, + boolean useContextSerdes) { + if ( useContextSerdes ) { + return RocksDBKeyValueStore.create("my-store", context); + } + return RocksDBKeyValueStore.create("my-store", context, keyClass, valueClass); + } +}