From b09def9cccd93ccebcd091fdca9796947d64ae6c Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Wed, 27 Jun 2018 11:25:22 -0400
Subject: [PATCH] Polishing Kafka Streams binder docs

Resolves #390
---
 .../src/main/asciidoc/kafka-streams.adoc      | 50 ++++++++++++++-----
 1 file changed, 37 insertions(+), 13 deletions(-)

diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
index b047aa787..74666b8b1 100644
--- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
+++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
@@ -539,7 +539,6 @@ handling yet.
 However, when you use the low-level Processor API in your application, there are options to control this behavior. See 
 below.
 
-
 [source]
 ----
 @Autowired
@@ -579,13 +578,13 @@ public KStream<?, WordCount> process(KStream<Object, String> input) {
 
 == 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.
+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.
 
-Creation code:
+Creating a state store:
 [source]
 ----
 @KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, lengthMs=300000)
@@ -594,7 +593,7 @@ public void process(KStream<Object, Product> input) {
 }
 ----
 
-Access code:
+Accessing the state store:
 [source]
 ----
 Processor<Object, Product>() {
@@ -611,14 +610,13 @@ Processor<Object, Product>() {
 
 == Interactive Queries
 
-As part of the public Kafka Streams binder API, we expose a class called `QueryableStoreRegistry`. You can access this 
-as a Spring bean in your application. An easy way to get access to this bean from your application is to "autowire" the bean 
-in your application.
+As part of the public Kafka Streams binder API, we expose a class called `InteractiveQueryService`.
+You can access this as a Spring bean in your application. An easy way to get access to this bean from your application is to "autowire" the bean.
 
 [source]
 ----
 @Autowired
-private QueryableStoreRegistry queryableStoreRegistry;
+private InteractiveQueryService interactiveQueryService;
 ----
 
 Once you gain access to this bean, then you can query for the particular state-store that you are interested. See below.
@@ -626,5 +624,31 @@ Once you gain access to this bean, then you can query for the particular state-s
 [source]
 ----
 ReadOnlyKeyValueStore<Object, Object> keyValueStore =
-						queryableStoreRegistry.getQueryableStoreType("my-store", QueryableStoreTypes.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
+}
 ----