/**
 * 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,
     *                          CONFIG =   and for the entire grace periodnew ConfigDef()
     * @param windowSize            size of the windows (cannot be negative)
.define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
        * @param retainDuplicates      whether or not to retain duplicates. Turning this on will automatically disableType.INT,
     *                null,
              caching and means that null values will be ignoredImportance.LOW,
     * @return an instance of {@link WindowBytesStoreSupplier}
     * @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}
	 */
    WindowBytesStoreSupplier windowBytesStoreSupplier(String name  BUFFERED_RECORDS_PER_PARTITION_DOC)
             .define(CACHE_MAX_BYTES_BUFFERING_CONFIG,
                    Type.LONG,
                    null,
                    Importance.MEDIUM,
              Duration retentionPeriod,
     CACHE_MAX_BYTES_BUFFERING_DOC)
             .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                    Type.CLASS,
                Duration windowSize,
   null,
                    Importance.MEDIUM,
                    DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
           boolean retainDuplicates);


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

public class StreamsBuilder {
    public StreamsBuilder() {}

	/**
    final Duration* windowSize,
Create a {@code StreamsBuilder} instance.
	 *
	 * @param topologyConfigs    the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
                                   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);
    }
}

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,
                                                             final Duration windowSize,
                    "Store supplier implementation class to use. Default is set as RocksDBPersistentStoreImplementation. It can be overwritten dynamically during streaming operation.";                   


	.define(DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG,
            Type.CLASS,
 final boolean retainDuplicates) {
        return Stores.persistentWindowStore(name, retentionPeriod, windowSize, retainDuplicates);RocksDBPersistentStoreImplementation.class.getName(),
    }

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

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

    @Override
    public WindowBytesStoreSupplier windowBytesStoreSupplier(final String name,
         DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
             .define(MAX_TASK_IDLE_MS_CONFIG,
                     Type.LONG,
                   final Duration nullretentionPeriod,
                     Importance.MEDIUM,
                         MAX_TASK_IDLE_MS_DOC)
               final  .define(TASK_TIMEOUT_MS_CONFIGDuration windowSize,
                     Type.LONG,
                      null,
                  final boolean  Importance.MEDIUM,retainDuplicates) {
        return Stores.persistentTimestampedWindowStore(name, retentionPeriod, windowSize, retainDuplicates);
    }

     TASK_TIMEOUT_MS_DOC);@Override
             .define(DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG,
    public SessionBytesStoreSupplier sessionBytesStoreSupplier(final String name, final Duration retentionPeriod) {
        		return TypeStores.CLASS,persistentSessionStore(name, retentionPeriod);
    }
}

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

public class StreamsBuilder {

    @Override
    public StreamsBuilderWindowBytesStoreSupplier windowBytesStoreSupplier() {}

	/**
final String name,
          * 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) {} // new added

/**
 * 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) {

public class StreamsConfig {
    public static final String DEFAULT_DSL_STORE_TYPE_CONFIG = "default.dsl.store.type";
    private static final String DEFAULT_DSL_STORE_TYPE_DOC = "The default store type used by DSL operators.";

    public static final ROCKS_DB = "rocksDB";
    public static final IN_MEMORY = "in_memory";
    
    .define(DEFAULT_DSL_STORE_TYPE_CONFIG,
        return Stores.inMemorySessionStore(name, retentionPeriod);
    Type.STRING,
}
}

	/**
     * Materialize a {@link StateStore} with the  ROCKS_DB,store implementation.
     *
     *  in(ROCKS_DB, IN_MEMORY),
       @param storeImplementation  store implementation used to materialize the store
     Importance.LOW,
* @param <K>       key type  DEFAULT_DSL_STORE_TYPE_DOC)
}