/**
 * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
 * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
 * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
 * topology builders via the {@link org.apache.kafka.streams.StreamsBuilder()} method.
 */
public class TopologyConfig extends AbstractConfig {
    private static final ConfigDef CONFIG;
    static {
                         (note that the retention period must be at least long enough to contain the
     *                              windowed data's entire life cycle, from window-start through window-end,
     *                              and for the entire grace period)
     * @param windowSize            size ofCONFIG the= windowsnew ConfigDef(cannot be negative)
     * @param retainDuplicates      whether or not to retain duplicates. Turning this on will automatically disable
.define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
          *           Type.INT,
                   caching and meansnull,
 that null values will be ignored.
     * @return an instance of {@link WindowBytesStoreSupplier}
    Importance.LOW,
 * @throws IllegalArgumentException if {@code retentionPeriod} or {@code windowSize} can't be represented as {@code long milliseconds}
     * @throws IllegalArgumentException if {@code retentionPeriod} is smaller than {@code windowSize}
	 */
BUFFERED_RECORDS_PER_PARTITION_DOC)
            WindowBytesStoreSupplier windowBytesStoreSupplier(String name.define(CACHE_MAX_BYTES_BUFFERING_CONFIG,
                    Type.LONG,
                    null,
              Duration retentionPeriod,
      Importance.MEDIUM,
                    CACHE_MAX_BYTES_BUFFERING_DOC)
             .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                   Duration windowSizeType.CLASS,
                    null,
                    Importance.MEDIUM,
               boolean retainDuplicates);


    /**DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
     * Create a {@link SessionBytesStoreSupplier}.
     * 
     * @param name.define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
                  name of the store (cannot be {@code null}) Type.CLASS,
     * @param retentionPeriod   length of time to retain data in the store (cannot be negative)null,
     *                 Importance.MEDIUM,
         (note that the retention period must be at least as long enough to DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
     *        .define(MAX_TASK_IDLE_MS_CONFIG,
                  contain the inactivity gapType.LONG,
 of the session and the entire grace period.)
     * @return an instance of a {@link  SessionBytesStoreSupplier}
null,
             */
      SessionBytesStoreSupplier sessionBytesStoreSupplier(String name Importance.MEDIUM,
                     MAX_TASK_IDLE_MS_DOC)
             .define(TASK_TIMEOUT_MS_CONFIG,
                     Type.LONG,
           Duration retentionPeriod);
}

public class StreamsBuilder {
    public StreamsBuilder() {}

	/**
     * Create a {@code StreamsBuilder} instance.
	 *
	 * @param topologyConfigs    the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
     */
	public StreamsBuilder(final TopologyConfig topologyConfigs) final{} boolean// retainDuplicates)new {
        return Stores.persistentTimestampedWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

    @Override
    public SessionBytesStoreSupplier sessionBytesStoreSupplier(final String name, final Duration retentionPeriod) {
        return Stores.persistentSessionStore(name, retentionPeriod);
    }
}

Properties props = new Properties();
...
// set the "default.dsl.store" config as "in-memory store"
props.put(StreamsConfig.DEFAULT_DSL_STORE_CONFIG, StreamsConfig.IN_MEMORY);
StreamsConfig streamsConfig = new StreamsConfig(props);

builder = new StreamsBuilder(new TopologyConfig(streamsConfig));
topology = builder.build(); 

KStream stream = ...

// current code to change from RocksDB to in-memory
stream.groupByKey().count(Materialized.as(Stores.inMemoryKeyValueStore("some-name"));

// new code, without the need to provide a name
stream.groupByKey().count(Materialized.withStoreType(StoreImplType.IN_MEMORY));

public class StreamsConfig {     
    public static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG = "default.store.impl.class";
    private static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_DOC = 

/**
 * A Rocks DB state store supplier Implementation.
 *
 */
public class RocksDBStoreImplementation implements StoreImplementation {
    @Override
    public KeyValueBytesStoreSupplier keyValueSupplier(final String name) {
        return Stores.persistentKeyValueStore(name);
    }

    @Override
    public WindowBytesStoreSupplier windowBytesStoreSupplier(final String name,
                                                             final Duration retentionPeriod,
                                   "Store supplier implementation class to use. Default is set as RocksDBPersistentStoreImplementation. It can be overwritten dynamically during streaming operation.";              final Duration windowSize   


	.define(DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG,
            Type.CLASS,
            RocksDBPersistentStoreImplementation.class.getName(),
            Importance.MEDIUM,
            DEFAULT_STORE_IMPLEMENTATION_CLASS_DOC) 
}

/**
 * A state store supplier Implementation interface for all types of final boolean retainDuplicates) {
        return Stores.persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

    @Override{@link StateStore}.
 *
 */
public interface StoreImplementation {
    /**
     * Create a {@link KeyValueBytesStoreSupplier}.
     *
    public * SessionBytesStoreSupplier@param sessionBytesStoreSupplier(finalname String name, final Duration retentionPeriod) {of the store (cannot be {@code null})
     * @return an return Stores.persistentSessionStore(name, retentionPeriod);
    }
}

/**
 * A Rocks DB persistent state store supplier Implementation.
 *
 */
public class RocksDBPersistentStoreImplementation implements StoreImplementation {
    @Override
    public KeyValueBytesStoreSupplier keyValueSupplier(final String name) {
        return Stores.persistentTimestampedKeyValueStore(name);
    }

    @Override
    public WindowBytesStoreSupplier windowBytesStoreSupplier(final String name,
                                                             final Duration retentionPeriod,
                                                             final Duration windowSize,
                                                             final boolean retainDuplicates) {
        return Stores.persistentTimestampedWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

    @Override
    public SessionBytesStoreSupplier sessionBytesStoreSupplier(final String name, final Duration retentionPeriod) {
        return Stores.persistentSessionStore(name, retentionPeriod);
    }
}

/**
 * A Rocks DB state store supplier Implementation.
 *
 */
public class RocksDBStoreImplementation implements StoreImplementation {
    @Override
    public KeyValueBytesStoreSupplier keyValueSupplier(final String name) {
        return Stores.persistentKeyValueStore(name);
    }

    @Override
    public WindowBytesStoreSupplier windowBytesStoreSupplier(final String name,
                                                             final Duration retentionPeriod,
                                                             final Duration windowSize,
                                                             final boolean retainDuplicates) {
        return Stores.persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

    @Override
    public SessionBytesStoreSupplier sessionBytesStoreSupplier(final String name, final Duration retentionPeriod) {
        return Stores.persistentSessionStore(name, retentionPeriod);
    }
}

/**
 * A In-memory state store supplier Implementation.
 *
 */
public class InMemoryStoreImplementation implements StoreImplementation {
    @Override
    public KeyValueBytesStoreSupplier keyValueSupplier(final String name) {
        return Stores.inMemoryKeyValueStore(name);
    }

    @Override
    public WindowBytesStoreSupplier windowBytesStoreSupplier(final String name,
                                                             final Duration retentionPeriod,
                                                             final Duration windowSize,
                                                             final boolean retainDuplicates) {
        return Stores.inMemoryWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

    @Override
    public SessionBytesStoreSupplier sessionBytesStoreSupplier(final String name, final Duration retentionPeriod) {
        return Stores.inMemorySessionStore(name, retentionPeriod);
    }
}

	/**
     * Materialize a {@link StateStore} with the store implementation.
     *
     * @param storeImplementation  store implementation used to materialize the store
     * @param <K>       key type of the store
     * @param <V>       value type of the store
     * @param <S>       type of the {@link StateStore}

     * @return a new {@link Materialized} instance with the given storeImplementation
     */
    public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreImplementation storeImplementation)