[WIP] demonstrating my thoughts on #6192

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/kstream/internals/KGroupedStreamImpl.java

@@ -185,10 +185,11 @@ private <T> KTable<K, T> doAggregate(final KStreamAggProcessorSupplier<K, K, V,
                                          final MaterializedInternal<K, T, KeyValueStore<Bytes, byte[]>> materializedInternal) {
         return aggregateBuilder.build(
             functionName,
-            new KeyValueStoreMaterializer<>(materializedInternal).materialize(),
+            new KeyValueWithTimestampStoreMaterializer<>(materializedInternal).materialize(),
             aggregateSupplier,
             materializedInternal.queryableStoreName(),
             materializedInternal.keySerde(),
-            materializedInternal.valueSerde());
+            materializedInternal.valueSerde()
+        );
     }
 }

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

@@ -82,7 +82,7 @@ private <T> KTable<K, T> doAggregate(final ProcessorSupplier<K, Change<V>> aggre
         final StatefulProcessorNode statefulProcessorNode = new StatefulProcessorNode<>(
             funcName,
             new ProcessorParameters<>(aggregateSupplier, funcName),
-            new KeyValueStoreMaterializer<>(materialized).materialize(),
+            new KeyValueWithTimestampStoreMaterializer<>(materialized).materialize(),
             false
         );
 

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

@@ -21,8 +21,12 @@
 import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.processor.Processor;
 import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
 import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+import org.apache.kafka.streams.state.internals.WrappedStateStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +57,7 @@ public void enableSendingOldValues() {
 
     private class KStreamAggregateProcessor extends AbstractProcessor<K, V> {
 
-        private KeyValueStore<K, T> store;
+        private KeyValueStore<K, ValueAndTimestamp<T>> store;
         private StreamsMetricsImpl metrics;
         private TupleForwarder<K, T> tupleForwarder;
 
@@ -62,11 +66,14 @@ private class KStreamAggregateProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            store = (KeyValueStore<K, T>) context.getStateStore(storeName);
+            StateStore store = context.getStateStore(storeName);
+            if (store instanceof WrappedStateStore) {
+                store = ((WrappedStateStore) store).wrappedStore();
+            }
+            this.store = ((KeyValueWithTimestampStoreMaterializer.TimestampHidingKeyValueStoreFacade) store).inner;
             tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<K, V>(context), sendOldValues);
         }
 
-
         @Override
         public void process(final K key, final V value) {
             // If the key or value is null we don't need to proceed
@@ -79,23 +86,28 @@ key, value, context().topic(), context().partition(), context().offset()
                 return;
             }
 
-            T oldAgg = store.get(key);
+            final ValueAndTimestamp<T> oldAggWithTimestamp = store.get(key);
+            final T oldAgg;
+            final long oldTimestamp;
 
-            if (oldAgg == null) {
+            if (oldAggWithTimestamp == null) {
                 oldAgg = initializer.apply();
+                oldTimestamp = -1;
+            } else {
+                oldAgg = oldAggWithTimestamp.value();
+                oldTimestamp = oldAggWithTimestamp.timestamp();
             }
 
-            T newAgg = oldAgg;
-
-            // try to add the new value
-            newAgg = aggregator.apply(key, value, newAgg);
+            final T newAgg = aggregator.apply(key, value, oldAgg);
+            final long newTimestamp = Math.max(oldTimestamp, context().timestamp());
 
             // update the store with the new value
-            store.put(key, newAgg);
-            tupleForwarder.maybeForward(key, newAgg, sendOldValues ? oldAgg : null);
+            store.put(key, ValueAndTimestamp.make(newAgg, newTimestamp));
+            tupleForwarder.maybeForward(key, newAgg, sendOldValues ? oldAgg : null, newTimestamp);
         }
     }
 
+
     @Override
     public KTableValueGetterSupplier<K, T> view() {
 
@@ -114,16 +126,16 @@ public String[] storeNames() {
 
     private class KStreamAggregateValueGetter implements KTableValueGetter<K, T> {
 
-        private KeyValueStore<K, T> store;
+        private ReadOnlyKeyValueStore<K, ValueAndTimestamp<T>> store;
 
         @SuppressWarnings("unchecked")
         @Override
         public void init(final ProcessorContext context) {
-            store = (KeyValueStore<K, T>) context.getStateStore(storeName);
+            store = (ReadOnlyKeyValueStore<K, ValueAndTimestamp<T>>) context.getStateStore(storeName);
         }
 
         @Override
-        public T get(final K key) {
+        public ValueAndTimestamp<T> get(final K key) {
             return store.get(key);
         }
 

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.kstream.internals;
 
 import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.ForeachAction;
 import org.apache.kafka.streams.kstream.GlobalKTable;
@@ -36,6 +37,7 @@
 import org.apache.kafka.streams.kstream.ValueMapperWithKey;
 import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
 import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode;
 import org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.OptimizableRepartitionNodeBuilder;
 import org.apache.kafka.streams.kstream.internals.graph.ProcessorGraphNode;
@@ -46,18 +48,28 @@
 import org.apache.kafka.streams.kstream.internals.graph.StreamTableJoinNode;
 import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
 import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
+import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
+import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.TopicNameExtractor;
 import org.apache.kafka.streams.processor.internals.StaticTopicNameExtractor;
+import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
 import org.apache.kafka.streams.state.WindowStore;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+import org.apache.kafka.streams.state.internals.KeyValueIteratorFacade;
+import org.apache.kafka.streams.state.internals.WindowStoreBuilder;
 
 import java.lang.reflect.Array;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
@@ -854,16 +866,208 @@ private static <K, V> StoreBuilder<WindowStore<K, V>> joinWindowStoreBuilder(fin
                                                                                  final JoinWindows windows,
                                                                                  final Serde<K> keySerde,
                                                                                  final Serde<V> valueSerde) {
-        return Stores.windowStoreBuilder(
-            Stores.persistentWindowStore(
-                joinName + "-store",
-                Duration.ofMillis(windows.size() + windows.gracePeriodMs()),
-                Duration.ofMillis(windows.size()),
-                true
-            ),
-            keySerde,
-            valueSerde
+        final WindowBytesStoreSupplier supplier = Stores.persistentWindowWithTimestampStore(
+            joinName + "-store",
+            Duration.ofMillis(windows.size() + windows.gracePeriodMs()),
+            Duration.ofMillis(windows.size()),
+            true
         );
+        return new WindowStoreBuilder<K, V>(supplier, null, null, Time.SYSTEM) {
+            final StoreBuilder<WindowStore<K, ValueAndTimestamp<V>>> inner = Stores.windowWithTimestampStoreBuilder(
+                supplier,
+                keySerde,
+                valueSerde
+            );
+
+            @Override
+            public WindowStoreBuilder<K, V> withCachingEnabled() {
+                inner.withCachingEnabled();
+                return this;
+            }
+
+            @Override
+            public WindowStoreBuilder<K, V> withCachingDisabled() {
+                inner.withCachingDisabled();
+                return this;
+            }
+
+            @Override
+            public WindowStoreBuilder<K, V> withLoggingEnabled(final Map<String, String> config) {
+                inner.withLoggingEnabled(config);
+                return this;
+            }
+
+            @Override
+            public WindowStoreBuilder<K, V> withLoggingDisabled() {
+                inner.withLoggingDisabled();
+                return this;
+            }
+
+            @Override
+            public Map<String, String> logConfig() {
+                return inner.logConfig();
+            }
+
+            @Override
+            public boolean loggingEnabled() {
+                return inner.loggingEnabled();
+            }
+
+            @Override
+            public String name() {
+                return inner.name();
+            }
+
+            @Override
+            public WindowStore<K, V> build() {
+                return new WindowStoreFacade<>(inner.build());
+            }
+        };
+    }
+
+    public static class WindowStoreFacade<A, B> implements WindowStore<A, B> {
+        public final WindowStore<A, ValueAndTimestamp<B>> inner;
+
+        public WindowStoreFacade(final WindowStore<A, ValueAndTimestamp<B>> store) {
+            this.inner = store;
+        }
+
+        @Override
+        public void init(final ProcessorContext context,
+                         final StateStore root) {
+            inner.init(context, root);
+        }
+
+        @Override
+        public void put(final A key,
+                        final B value) {
+            inner.put(key, ValueAndTimestamp.make(value, -1L));
+        }
+
+        @Override
+        public void put(final A key,
+                        final B value,
+                        final long windowStartTimestamp) {
+            inner.put(key, ValueAndTimestamp.make(value, -1L), windowStartTimestamp);
+        }
+
+        @Override
+        public B fetch(final A key,
+                       final long time) {
+            final ValueAndTimestamp<B> valueAndTimestamp = inner.fetch(key, time);
+            return valueAndTimestamp == null ? null : valueAndTimestamp.value();
+        }
+
+        @Override
+        @SuppressWarnings("deprecation")
+        public WindowStoreIterator<B> fetch(final A key,
+                                            final long timeFrom,
+                                            final long timeTo) {
+            final KeyValueIterator<Long, ValueAndTimestamp<B>> innerIterator = inner.fetch(key, timeFrom, timeTo);
+            return new WindowStoreIteratorFacade(innerIterator);
+        }
+
+        @Override
+        public WindowStoreIterator<B> fetch(final A key,
+                                            final Instant from,
+                                            final Instant to) throws IllegalArgumentException {
+            final KeyValueIterator<Long, ValueAndTimestamp<B>> innerIterator = inner.fetch(key, from, to);
+            return new WindowStoreIteratorFacade(innerIterator);
+        }
+
+        @Override
+        @SuppressWarnings("deprecation")
+        public KeyValueIterator<Windowed<A>, B> fetch(final A from,
+                                                      final A to,
+                                                      final long timeFrom,
+                                                      final long timeTo) {
+            final KeyValueIterator<Windowed<A>, ValueAndTimestamp<B>> innerIterator = inner.fetch(from, to, timeFrom, timeTo);
+            return new KeyValueIteratorFacade<>(innerIterator);
+        }
+
+        @Override
+        public KeyValueIterator<Windowed<A>, B> fetch(final A from,
+                                                      final A to,
+                                                      final Instant fromTime,
+                                                      final Instant toTime) throws IllegalArgumentException {
+            final KeyValueIterator<Windowed<A>, ValueAndTimestamp<B>> innerIterator = inner.fetch(from, to, fromTime, toTime);
+            return new KeyValueIteratorFacade<>(innerIterator);
+        }
+
+        @Override
+        @SuppressWarnings("deprecation")
+        public KeyValueIterator<Windowed<A>, B> fetchAll(final long timeFrom,
+                                                         final long timeTo) {
+            final KeyValueIterator<Windowed<A>, ValueAndTimestamp<B>> innerIterator = inner.fetchAll(timeFrom, timeTo);
+            return new KeyValueIteratorFacade<>(innerIterator);
+        }
+
+        @Override
+        public KeyValueIterator<Windowed<A>, B> fetchAll(final Instant from,
+                                                         final Instant to) throws IllegalArgumentException {
+            final KeyValueIterator<Windowed<A>, ValueAndTimestamp<B>> innerIterator = inner.fetchAll(from, to);
+            return new KeyValueIteratorFacade<>(innerIterator);
+        }
+
+        @Override
+        public KeyValueIterator<Windowed<A>, B> all() {
+            final KeyValueIterator<Windowed<A>, ValueAndTimestamp<B>> innerIterator = inner.all();
+            return new KeyValueIteratorFacade<>(innerIterator);
+        }
+
+        @Override
+        public void flush() {
+            inner.flush();
+        }
+
+        @Override
+        public void close() {
+            inner.close();
+        }
+
+        @Override
+        public boolean isOpen() {
+            return inner.isOpen();
+        }
+
+        @Override
+        public String name() {
+            return inner.name();
+        }
+
+        @Override
+        public boolean persistent() {
+            return inner.persistent();
+        }
+
+        private class WindowStoreIteratorFacade implements WindowStoreIterator<B> {
+            final KeyValueIterator<Long, ValueAndTimestamp<B>> innerIterator;
+
+            WindowStoreIteratorFacade(final KeyValueIterator<Long, ValueAndTimestamp<B>> iterator) {
+                innerIterator = iterator;
+            }
+
+            @Override
+            public void close() {
+                innerIterator.close();
+            }
+
+            @Override
+            public Long peekNextKey() {
+                return innerIterator.peekNextKey();
+            }
+
+            @Override
+            public boolean hasNext() {
+                return innerIterator.hasNext();
+            }
+
+            @Override
+            public KeyValue<Long, B> next() {
+                final KeyValue<Long, ValueAndTimestamp<B>> innerKeyValue = innerIterator.next();
+                return KeyValue.pair(innerKeyValue.key, innerKeyValue.value.value());
+            }
+        }
     }
 
     private class KStreamImplJoin {