KAFKA-3522: Add TimestampedKeyValueStore builder/runtime classes

streams/src/main/java/org/apache/kafka/streams/kstream/internals/ForwardingCacheFlushListener.java

@@ -39,7 +39,7 @@ public void apply(final K key,
         final ProcessorNode prev = context.currentNode();
         context.setCurrentNode(myNode);
         try {
-            context.forward(key, new Change<>(newValue, oldValue),  To.all().withTimestamp(timestamp));
+            context.forward(key, new Change<>(newValue, oldValue), To.all().withTimestamp(timestamp));
         } finally {
             context.setCurrentNode(prev);
         }

streams/src/main/java/org/apache/kafka/streams/state/TimestampedKeyValueStore.java

@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+/**
+ * A key-(value/timestamp) store that supports put/get/delete and range queries.
+ *
+ * @param <K> The key type
+ * @param <V> The value type
+ */
+public interface TimestampedKeyValueStore<K, V> extends KeyValueStore<K, ValueAndTimestamp<V>> { }

streams/src/main/java/org/apache/kafka/streams/state/ValueAndTimestamp.java

@@ -0,0 +1,74 @@
+/*
+ * 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.KeyValue;
+
+import java.util.Objects;
+
+/**
+ * Combines a value from a {@link KeyValue} with a timestamp.
+ *
+ * @param <V>
+ */
+public class ValueAndTimestamp<V> {
+    private final V value;
+    private final long timestamp;
+
+    private ValueAndTimestamp(final V value,
+                              final long timestamp) {
+        Objects.requireNonNull(value);
+        this.value = value;
+        this.timestamp = timestamp;
+    }
+
+    public static <V> ValueAndTimestamp<V> make(final V value,
+                                                final long timestamp) {
+        return value == null ? null : new ValueAndTimestamp<>(value, timestamp);
+    }
+
+    public V value() {
+        return value;
+    }
+
+    public long timestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public String toString() {
+        return "<" + value + "," + timestamp + ">";
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final ValueAndTimestamp<?> that = (ValueAndTimestamp<?>) o;
+        return timestamp == that.timestamp &&
+            Objects.equals(value, that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(value, timestamp);
+    }
+}

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java

@@ -73,12 +73,12 @@ private void initInternal(final ProcessorContext context) {
     private void putAndMaybeForward(final ThreadCache.DirtyEntry entry,
                                     final InternalProcessorContext context) {
         if (flushListener != null) {
-            final byte[] newValueBytes = entry.newValue();
-            final byte[] oldValueBytes = newValueBytes == null || sendOldValues ? wrapped().get(entry.key()) : null;
+            final byte[] rawNewValue = entry.newValue();
+            final byte[] rawOldValue = rawNewValue == null || sendOldValues ? wrapped().get(entry.key()) : null;
 
             // this is an optimization: if this key did not exist in underlying store and also not in the cache,
             // we can skip flushing to downstream as well as writing to underlying store
-            if (newValueBytes != null || oldValueBytes != null) {
+            if (rawNewValue != null || rawOldValue != null) {
                 // we need to get the old values if needed, and then put to store, and then flush
                 wrapped().put(entry.key(), entry.newValue());
 
@@ -87,8 +87,8 @@ private void putAndMaybeForward(final ThreadCache.DirtyEntry entry,
                 try {
                     flushListener.apply(
                         entry.key().get(),
-                        newValueBytes,
-                        sendOldValues ? oldValueBytes : null,
+                        rawNewValue,
+                        sendOldValues ? rawOldValue : null,
                         entry.entry().context().timestamp());
                 } finally {
                     context.setRecordContext(current);
@@ -237,7 +237,8 @@ public KeyValueIterator<Bytes, byte[]> range(final Bytes from,
     @Override
     public KeyValueIterator<Bytes, byte[]> all() {
         validateStoreOpen();
-        final KeyValueIterator<Bytes, byte[]> storeIterator = new DelegatingPeekingKeyValueIterator<>(this.name(), wrapped().all());
+        final KeyValueIterator<Bytes, byte[]> storeIterator =
+            new DelegatingPeekingKeyValueIterator<>(this.name(), wrapped().all());
         final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.all(cacheName);
         return new MergedSortedCacheKeyValueBytesStoreIterator(cacheIterator, storeIterator);
     }

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java

@@ -143,7 +143,8 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes key,
                                                                              key,
                                                                              earliestSessionEndTime,
                                                                              latestSessionStartTime);
-        final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
+        final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
+            new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
         return new MergedSortedCacheSessionStoreIterator(filteredCacheIterator, storeIterator, cacheFunction);
     }
 
@@ -165,7 +166,8 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes keyFro
                                                                              keyTo,
                                                                              earliestSessionEndTime,
                                                                              latestSessionStartTime);
-        final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
+        final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
+            new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
         return new MergedSortedCacheSessionStoreIterator(filteredCacheIterator, storeIterator, cacheFunction);
     }
 
@@ -194,7 +196,8 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes key) {
     }
 
     @Override
-    public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from, final Bytes to) {
+    public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from,
+                                                           final Bytes to) {
         Objects.requireNonNull(from, "from cannot be null");
         Objects.requireNonNull(to, "to cannot be null");
         return findSessions(from, to, 0, Long.MAX_VALUE);

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java

@@ -82,31 +82,32 @@ private void putAndMaybeForward(final ThreadCache.DirtyEntry entry,
         final byte[] binaryWindowKey = cacheFunction.key(entry.key()).get();
         final Windowed<Bytes> windowedKeyBytes = WindowKeySchema.fromStoreBytesKey(binaryWindowKey, windowSize);
         final long windowStartTimestamp = windowedKeyBytes.window().start();
-        final Bytes key = windowedKeyBytes.key();
+        final Bytes binaryKey = windowedKeyBytes.key();
         if (flushListener != null) {
-            final byte[] newValueBytes = entry.newValue();
-            final byte[] oldValueBytes = newValueBytes == null || sendOldValues ? wrapped().fetch(key, windowStartTimestamp) : null;
+            final byte[] rawNewValue = entry.newValue();
+            final byte[] rawOldValue = rawNewValue == null || sendOldValues ?
+                wrapped().fetch(binaryKey, windowStartTimestamp) : null;
 
             // this is an optimization: if this key did not exist in underlying store and also not in the cache,
             // we can skip flushing to downstream as well as writing to underlying store
-            if (newValueBytes != null || oldValueBytes != null) {
+            if (rawNewValue != null || rawOldValue != null) {
                 // we need to get the old values if needed, and then put to store, and then flush
-                wrapped().put(key, entry.newValue(), windowStartTimestamp);
+                wrapped().put(binaryKey, entry.newValue(), windowStartTimestamp);
 
                 final ProcessorRecordContext current = context.recordContext();
                 context.setRecordContext(entry.entry().context());
                 try {
                     flushListener.apply(
                         binaryWindowKey,
-                        newValueBytes,
-                        sendOldValues ? oldValueBytes : null,
+                        rawNewValue,
+                        sendOldValues ? rawOldValue : null,
                         entry.entry().context().timestamp());
                 } finally {
                     context.setRecordContext(current);
                 }
             }
         } else {
-            wrapped().put(key, entry.newValue(), windowStartTimestamp);
+            wrapped().put(binaryKey, entry.newValue(), windowStartTimestamp);
         }
     }
 

streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java

@@ -32,7 +32,7 @@ public class ChangeLoggingKeyValueBytesStore
     extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, byte[], byte[]>
     implements KeyValueStore<Bytes, byte[]> {
 
-    private StoreChangeLogger<Bytes, byte[]> changeLogger;
+    StoreChangeLogger<Bytes, byte[]> changeLogger;
 
     ChangeLoggingKeyValueBytesStore(final KeyValueStore<Bytes, byte[]> inner) {
         super(inner);
@@ -43,13 +43,16 @@ public void init(final ProcessorContext context,
                      final StateStore root) {
         super.init(context, root);
         final String topic = ProcessorStateManager.storeChangelogTopic(context.applicationId(), name());
-        changeLogger = new StoreChangeLogger<>(name(), context, new StateSerdes<>(topic, Serdes.Bytes(), Serdes.ByteArray()));
+        changeLogger = new StoreChangeLogger<>(
+            name(),
+            context,
+            new StateSerdes<>(topic, Serdes.Bytes(), Serdes.ByteArray()));
 
         // if the inner store is an LRU cache, add the eviction listener to log removed record
         if (wrapped() instanceof MemoryLRUCache) {
             ((MemoryLRUCache) wrapped()).setWhenEldestRemoved((key, value) -> {
                 // pass null to indicate removal
-                changeLogger.logChange(key, null);
+                log(key, null);
             });
         }
     }
@@ -63,7 +66,7 @@ public long approximateNumEntries() {
     public void put(final Bytes key,
                     final byte[] value) {
         wrapped().put(key, value);
-        changeLogger.logChange(key, value);
+        log(key, value);
     }
 
     @Override
@@ -72,7 +75,7 @@ public byte[] putIfAbsent(final Bytes key,
         final byte[] previous = wrapped().putIfAbsent(key, value);
         if (previous == null) {
             // then it was absent
-            changeLogger.logChange(key, value);
+            log(key, value);
         }
         return previous;
     }
@@ -81,14 +84,14 @@ public byte[] putIfAbsent(final Bytes key,
     public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
         wrapped().putAll(entries);
         for (final KeyValue<Bytes, byte[]> entry : entries) {
-            changeLogger.logChange(entry.key, entry.value);
+            log(entry.key, entry.value);
         }
     }
 
     @Override
     public byte[] delete(final Bytes key) {
         final byte[] oldValue = wrapped().delete(key);
-        changeLogger.logChange(key, null);
+        log(key, null);
         return oldValue;
     }
 
@@ -107,4 +110,9 @@ public KeyValueIterator<Bytes, byte[]> range(final Bytes from,
     public KeyValueIterator<Bytes, byte[]> all() {
         return wrapped().all();
     }
+
+    void log(final Bytes key,
+             final byte[] value) {
+        changeLogger.logChange(key, value);
+    }
 }

streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStore.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.state.KeyValueStore;
+
+import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.rawValue;
+import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.timestamp;
+
+public class ChangeLoggingTimestampedKeyValueBytesStore extends ChangeLoggingKeyValueBytesStore {
+    ChangeLoggingTimestampedKeyValueBytesStore(final KeyValueStore<Bytes, byte[]> inner) {
+        super(inner);
+    }
+
+    @Override
+    void log(final Bytes key,
+             final byte[] valueAndTimestamp) {
+        if (valueAndTimestamp != null) {
+            changeLogger.logChange(key, rawValue(valueAndTimestamp), timestamp(valueAndTimestamp));
+        } else {
+            changeLogger.logChange(key, null);
+        }
+    }
+}