anchor.diff:  
   == State Store
   
  -State store is created automatically by Kafka Stream when Streas DSL is used. When use processor API, in case you want to
  -create and register a state store manually, you can use `KafkaStreamsStateStore` annotation. You can specify store name,
  -type, whether to enable log, whether disable cache, etc, and those parameters will be injected into KStream building
  -process in Kafka Streams binder to create and register the store to your KStream. After that, you can access the same way
  -how you access in normal Kafka Streams code.
  -
  -Creation code:
  +State store is created automatically by Kafka Streams when the DSL is used.
  +When processor API is used, you need to register a state store manually. In order to do so, you can use `KafkaStreamsStateStore` annotation.
  +You can specify the name and type of the store, flags to control log and disabling cache, etc.
  +Once the store is created by the binder during the bootstrapping phase, you can access this state store through the processor API.
  +Below are some primitives for doing this.
  +
  +Creating a state store:
   [source]
   ----
   @KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs=300000)
  
positive.code: However, when you use the low-level Processor API in your application, there are options to control this behavior. See 
  below.
  
  
  [source]
  ----
  @Autowired
  
negative.code: [source]
  ----
  ReadOnlyKeyValueStore<Object, Object> keyValueStore =
  						interactiveQueryService.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
  ----
  
  If there are multiple instances of the kafka streams application running, then before you can query them interactively, you need to identify which application instance hosts the key.
  `InteractiveQueryService` API provides methods for identifying the host information.
  
  In order for this to work, you must configure the property `application.server` as below:
  
  [source]
  ----
  spring.cloud.stream.kafka.streams.binder.configuration.application.server: <server>:<port>
  ----
  
  Here are some code snippets:
  
  [source]
  ----
  org.apache.kafka.streams.state.HostInfo hostInfo = interactiveQueryService.getHostInfo("store-name",
  						key, keySerializer);
  
  if (interactiveQueryService.getCurrentHostInfo().equals(hostInfo)) {
  
      //query from the store that is locally available
  }
  else {
      //query from the remote host
  }
  ----
  