Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/state/BuiltInDslStoreSuppliers.java (Java)

Code before:
 */
package org.apache.kafka.streams.state;

import org.apache.kafka.streams.DslStoreFormat;
import org.apache.kafka.streams.kstream.EmitStrategy;
import org.apache.kafka.streams.state.internals.RocksDbIndexedTimeOrderedWindowBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDbTimeOrderedSessionBytesStoreSupplier;

/**
 * Collection of builtin {@link DslStoreSuppliers} for Kafka Streams. Today we
 * support RocksDb and InMemory stores out of the box.
 */
public class BuiltInDslStoreSuppliers {

    public static final DslStoreSuppliers ROCKS_DB = new RocksDBDslStoreSuppliers();
    public static final DslStoreSuppliers IN_MEMORY = new InMemoryDslStoreSuppliers();

    /**
     * A {@link DslStoreSuppliers} that supplies all stores backed by RocksDB
     */
    public static class RocksDBDslStoreSuppliers implements DslStoreSuppliers {

        @Override
        public KeyValueBytesStoreSupplier keyValueStore(final DslKeyValueParams params) {
            final DslStoreFormat storeFormat = params.dslStoreFormat();
            if (storeFormat.equals(DslStoreFormat.HEADERS)) {
                return Stores.persistentTimestampedKeyValueStoreWithHeaders(params.name());
            }
            if (storeFormat.equals(DslStoreFormat.TIMESTAMPED)) {
                return Stores.persistentTimestampedKeyValueStore(params.name());
            }
            if (storeFormat.equals(DslStoreFormat.PLAIN)) {
                return Stores.persistentKeyValueStore(params.name());
            }
            throw new IllegalArgumentException(
                "Unsupported DslStoreFormat: " + storeFormat +
                    ". Expected one of: HEADERS, TIMESTAMPED, or PLAIN"
            );
        }

        @Override
        public WindowBytesStoreSupplier windowStore(final DslWindowParams params) {
            if (params.emitStrategy().type() == EmitStrategy.StrategyType.ON_WINDOW_CLOSE) {
                return RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.create(
                        params.name(),
                        params.retentionPeriod(),
                        params.windowSize(),
                        params.retainDuplicates(),
                        params.isSlidingWindow());
            }


Review comment (suggestion):
```suggestion
            if (storeFormat == DslStoreFormat.HEADERS) {
```

Code after:
 */
package org.apache.kafka.streams.state;

import org.apache.kafka.streams.DslStoreFormat;
import org.apache.kafka.streams.kstream.EmitStrategy;
import org.apache.kafka.streams.state.internals.RocksDbIndexedTimeOrderedWindowBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDbTimeOrderedSessionBytesStoreSupplier;

/**
 * Collection of builtin {@link DslStoreSuppliers} for Kafka Streams. Today we
 * support RocksDb and InMemory stores out of the box.
 */
public class BuiltInDslStoreSuppliers {

    public static final DslStoreSuppliers ROCKS_DB = new RocksDBDslStoreSuppliers();
    public static final DslStoreSuppliers IN_MEMORY = new InMemoryDslStoreSuppliers();

    /**
     * A {@link DslStoreSuppliers} that supplies all stores backed by RocksDB
     */
    public static class RocksDBDslStoreSuppliers implements DslStoreSuppliers {

        @Override
        public KeyValueBytesStoreSupplier keyValueStore(final DslKeyValueParams params) {
            final DslStoreFormat storeFormat = params.dslStoreFormat();
            switch (storeFormat) {
                case HEADERS:
                    return Stores.persistentTimestampedKeyValueStoreWithHeaders(params.name());
                case TIMESTAMPED:
                    return Stores.persistentTimestampedKeyValueStore(params.name());
                case PLAIN:
                    return Stores.persistentKeyValueStore(params.name());
                default:
                    throw new IllegalArgumentException("Unsupported DslStoreFormat: " + storeFormat +
                        ". Expected one of: HEADERS, TIMESTAMPED, or PLAIN");
            }
        }

        @Override
        public WindowBytesStoreSupplier windowStore(final DslWindowParams params) {
            if (params.emitStrategy().type() == EmitStrategy.StrategyType.ON_WINDOW_CLOSE) {
                return RocksDbIndexedTimeOrderedWindowBytesStoreSupplier.create(
                        params.name(),
                        params.retentionPeriod(),
                        params.windowSize(),
                        params.retainDuplicates(),
                        params.isSlidingWindow());
            }

            if (params.isTimestamped()) {
                return Stores.persistentTimestampedWindowStore(