KAFKA-20194: Ensure backward compatibility for Windowed Store

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
+            );
+        }
     }
 
     /**

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(
@@ -78,16 +78,6 @@ public WindowStore<Bytes, byte[]> 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);
         }

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

@@ -0,0 +1,58 @@
+/*
+ * 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.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<Bytes, byte[]> get() {
+        return new RocksDBTimestampedWindowStoreWithHeaders(
+            new RocksDBTimestampedSegmentedBytesStoreWithHeaders(
+                name,
+                metricsScope(),
+                retentionPeriod,
+                segmentInterval,
+                new WindowKeySchema()),
+            retainDuplicates,
+            windowSize);
+    }
+}