From 28f27266edbd9e98cc3189abb0f0e14887a5c1f1 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Wed, 8 Apr 2026 13:51:53 -0700 Subject: [PATCH 1/3] KAFKA-20194: Ensure backward compatibility for Windowed Store WindowHeader-supplier must implement HeadersBytesStoreSupplier. --- .../apache/kafka/streams/state/Stores.java | 30 +++++++--- .../RocksDbWindowBytesStoreSupplier.java | 10 ++-- ...ocksDbWindowHeadersBytesStoreSupplier.java | 59 +++++++++++++++++++ 3 files changed, 85 insertions(+), 14 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java index dfa2a2e2bd0bd..675cf06d45c5a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java @@ -30,6 +30,7 @@ import org.apache.kafka.streams.state.internals.RocksDbSessionHeadersBytesStoreSupplier; import org.apache.kafka.streams.state.internals.RocksDbVersionedKeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier; +import org.apache.kafka.streams.state.internals.RocksDbWindowHeadersBytesStoreSupplier; import org.apache.kafka.streams.state.internals.SessionStoreBuilder; import org.apache.kafka.streams.state.internals.SessionStoreWithHeadersBuilder; import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilder; @@ -44,6 +45,7 @@ import static org.apache.kafka.streams.internals.ApiUtils.prepareMillisCheckFailMsgPrefix; import static org.apache.kafka.streams.internals.ApiUtils.validateMillisecondDuration; +import static org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS; /** * Factory for creating state stores in Kafka Streams. @@ -360,7 +362,7 @@ public static WindowBytesStoreSupplier persistentTimestampedWindowStoreWithHeade final Duration windowSize, final boolean retainDuplicates ) throws IllegalArgumentException { - return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS); + return persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates, TIMESTAMPED_WINDOW_STORE_WITH_HEADERS); } private static WindowBytesStoreSupplier persistentWindowStore( @@ -390,14 +392,24 @@ private static WindowBytesStoreSupplier persistentWindowStore( + windowSize + "], retention=[" + retentionPeriod + "]"); } - return new RocksDbWindowBytesStoreSupplier( - name, - retentionMs, - defaultSegmentInterval, - windowSizeMs, - retainDuplicates, - storeType - ); + if (storeType == TIMESTAMPED_WINDOW_STORE_WITH_HEADERS) { + return new RocksDbWindowHeadersBytesStoreSupplier( + name, + retentionMs, + defaultSegmentInterval, + windowSizeMs, + retainDuplicates + ); + } else { + return new RocksDbWindowBytesStoreSupplier( + name, + retentionMs, + defaultSegmentInterval, + windowSizeMs, + retainDuplicates, + storeType + ); + } } /** diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java index 0d958d82822df..44f70811f327a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java @@ -27,11 +27,11 @@ public enum WindowStoreTypes { TIMESTAMPED_WINDOW_STORE_WITH_HEADERS } - private final String name; - private final long retentionPeriod; - private final long segmentInterval; - private final long windowSize; - private final boolean retainDuplicates; + protected final String name; + protected final long retentionPeriod; + protected final long segmentInterval; + protected final long windowSize; + protected final boolean retainDuplicates; private final WindowStoreTypes windowStoreType; public RocksDbWindowBytesStoreSupplier( diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java new file mode 100644 index 0000000000000..64b731ce91480 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java @@ -0,0 +1,59 @@ +/* + * 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.HeadersBytesStoreSupplier; +import org.apache.kafka.streams.state.WindowBytesStoreSupplier; +import org.apache.kafka.streams.state.WindowStore; + +import static org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS; + +public class RocksDbWindowHeadersBytesStoreSupplier + extends RocksDbWindowBytesStoreSupplier + implements HeadersBytesStoreSupplier { + + public RocksDbWindowHeadersBytesStoreSupplier( + final String name, + final long retentionPeriod, + final long segmentInterval, + final long windowSize, + final boolean retainDuplicates + ) { + super( + name, + retentionPeriod, + segmentInterval, + windowSize, + retainDuplicates, + TIMESTAMPED_WINDOW_STORE_WITH_HEADERS + ); + } + + @Override + public WindowStore get() { + return new RocksDBTimestampedWindowStoreWithHeaders( + new RocksDBTimestampedSegmentedBytesStoreWithHeaders( + name, + metricsScope(), + retentionPeriod, + segmentInterval, + new WindowKeySchema()), + retainDuplicates, + windowSize); + } +} From b99be8d436e0028ec0e956836b82cfe57a8ca788 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Wed, 8 Apr 2026 13:58:08 -0700 Subject: [PATCH 2/3] cleanup --- .../internals/RocksDbWindowBytesStoreSupplier.java | 13 +++---------- .../RocksDbWindowHeadersBytesStoreSupplier.java | 1 - 2 files changed, 3 insertions(+), 11 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java index 44f70811f327a..6fd69518da10d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java @@ -48,6 +48,9 @@ public RocksDbWindowBytesStoreSupplier( this.windowSize = windowSize; this.retainDuplicates = retainDuplicates; this.windowStoreType = windowStoreType; + if (windowStoreType == WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS) { + throw new IllegalArgumentException("This supplier does not support TIMESTAMPED_WINDOW_STORE_WITH_HEADERS type. Use RocksDbWindowHeadersBytesStoreSupplier instead."); + } } @Override @@ -78,16 +81,6 @@ public WindowStore get() { new WindowKeySchema()), retainDuplicates, windowSize); - case TIMESTAMPED_WINDOW_STORE_WITH_HEADERS: - return new RocksDBTimestampedWindowStoreWithHeaders( - new RocksDBTimestampedSegmentedBytesStoreWithHeaders( - name, - metricsScope(), - retentionPeriod, - segmentInterval, - new WindowKeySchema()), - retainDuplicates, - windowSize); default: throw new IllegalArgumentException("invalid window store type: " + windowStoreType); } diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java index 64b731ce91480..87384bcdc9a96 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowHeadersBytesStoreSupplier.java @@ -18,7 +18,6 @@ import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.state.HeadersBytesStoreSupplier; -import org.apache.kafka.streams.state.WindowBytesStoreSupplier; import org.apache.kafka.streams.state.WindowStore; import static org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier.WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS; From 1dec0317193cefd9bbfd2da9a56e2a9ebe6bd8a1 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Wed, 8 Apr 2026 18:24:22 -0700 Subject: [PATCH 3/3] remove too restrictve check --- .../state/internals/RocksDbWindowBytesStoreSupplier.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java index 6fd69518da10d..6a5ec29145b80 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbWindowBytesStoreSupplier.java @@ -48,9 +48,6 @@ public RocksDbWindowBytesStoreSupplier( this.windowSize = windowSize; this.retainDuplicates = retainDuplicates; this.windowStoreType = windowStoreType; - if (windowStoreType == WindowStoreTypes.TIMESTAMPED_WINDOW_STORE_WITH_HEADERS) { - throw new IllegalArgumentException("This supplier does not support TIMESTAMPED_WINDOW_STORE_WITH_HEADERS type. Use RocksDbWindowHeadersBytesStoreSupplier instead."); - } } @Override