<spanid="streams-developer-guide-interactive-queries"></span><h1>Interactive Queries<aclass="headerlink"href="#interactive-queries"title="Permalink to this headline"></a></h1>
<p>Interactive queries allow you to leverage the state of your application from outside your application. The Kafka Streams enables your applications to be queryable.</p>
<li><aclass="reference internal"href="#querying-local-state-stores-for-an-app-instance"id="id3">Querying local state stores for an app instance</a><ul>
<li><aclass="reference internal"href="#querying-local-key-value-stores"id="id4">Querying local key-value stores</a></li>
<li><aclass="reference internal"href="#querying-local-window-stores"id="id5">Querying local window stores</a></li>
<li><aclass="reference internal"href="#querying-local-custom-state-stores"id="id6">Querying local custom state stores</a></li>
</ul>
</li>
<li><aclass="reference internal"href="#querying-remote-state-stores-for-the-entire-app"id="id7">Querying remote state stores for the entire app</a><ul>
<li><aclass="reference internal"href="#adding-an-rpc-layer-to-your-application"id="id8">Adding an RPC layer to your application</a></li>
<li><aclass="reference internal"href="#exposing-the-rpc-endpoints-of-your-application"id="id9">Exposing the RPC endpoints of your application</a></li>
<li><aclass="reference internal"href="#discovering-and-accessing-application-instances-and-their-local-state-stores"id="id10">Discovering and accessing application instances and their local state stores</a></li>
<p>The full state of your application is typically <aclass="reference internal"href="../architecture.html#streams_architecture_state"><spanclass="std std-ref">split across many distributed instances of your application</span></a>, and across many state stores that are managed locally by these application instances.</p>
<p>There are local and remote components to interactively querying the state of your application.</p>
<dlclass="docutils">
<dt>Local state</dt>
<dd>An application instance can query the locally managed portion of the state and directly query its own local state stores. You can use the corresponding local data in other parts of your application code, as long as it doesn’t required calling the Kafka Streams API. Querying state stores is always read-only to guarantee that the underlying state stores will never be mutated out-of-band (e.g., you cannot add new entries). State stores should only be mutated by the corresponding processor topology and the input data it operates on. For more information, see <aclass="reference internal"href="#streams-developer-guide-interactive-queries-local-stores"><spanclass="std std-ref">Querying local state stores for an app instance</span></a>.</dd>
<dt>Remote state</dt>
<dd><pclass="first">To query the full state of your application, you must connect the various fragments of the state, including:</p>
<ulclass="simple">
<li>query local state stores</li>
<li>discover all running instances of your application in the network and their state stores</li>
<li>communicate with these instances over the network (e.g., an RPC layer)</li>
</ul>
<pclass="last">Connecting these fragments enables communication between instances of the same app and communication from other applications for interactive queries. For more information, see <aclass="reference internal"href="#streams-developer-guide-interactive-queries-discovery"><spanclass="std std-ref">Querying remote state stores for the entire app</span></a>.</p>
</dd>
</dl>
<p>Kafka Streams natively provides all of the required functionality for interactively querying the state of your application, except if you want to expose the full state of your application via interactive queries. To allow application instances to communicate over the network, you must add a Remote Procedure Call (RPC) layer to your application (e.g., REST API).</p>
<p>This table shows the Kafka Streams native communication support for various procedures.</p>
<tableborder="1"class="docutils">
<colgroup>
<colwidth="42%"/>
<colwidth="27%"/>
<colwidth="31%"/>
</colgroup>
<theadvalign="bottom">
<trclass="row-odd"><thclass="head">Procedure</th>
<thclass="head">Application instance</th>
<thclass="head">Entire application</th>
</tr>
</thead>
<tbodyvalign="top">
<trclass="row-even"><td>Query local state stores of an app instance</td>
<td>Supported</td>
<td>Supported</td>
</tr>
<trclass="row-odd"><td>Make an app instance discoverable to others</td>
<td>Supported</td>
<td>Supported</td>
</tr>
<trclass="row-even"><td>Discover all running app instances and their state stores</td>
<td>Supported</td>
<td>Supported</td>
</tr>
<trclass="row-odd"><td>Communicate with app instances over the network (RPC)</td>
<spanid="streams-developer-guide-interactive-queries-local-stores"></span><h2><aclass="toc-backref"href="#id3">Querying local state stores for an app instance</a><aclass="headerlink"href="#querying-local-state-stores-for-an-app-instance"title="Permalink to this headline"></a></h2>
<p>A Kafka Streams application typically runs on multiple instances. The state that is locally available on any given instance is only a subset of the <aclass="reference internal"href="../architecture.html#streams-architecture-state"><spanclass="std std-ref">application’s entire state</span></a>. Querying the local stores on an instance will only return data locally available on that particular instance.</p>
<p>The method <codeclass="docutils literal"><spanclass="pre">KafkaStreams#store(...)</span></code> finds an application instance’s local state stores by name and type.</p>
<pclass="caption"><spanclass="caption-text">Every application instance can directly query any of its local state stores.</span></p>
</div>
<p>The <em>name</em> of a state store is defined when you create the store. You can create the store explicitly by using the Processor API or implicitly by using stateful operations in the DSL.</p>
<p>The <em>type</em> of a state store is defined by <codeclass="docutils literal"><spanclass="pre">QueryableStoreType</span></code>. You can access the built-in types via the class <codeclass="docutils literal"><spanclass="pre">QueryableStoreTypes</span></code>.
Kafka Streams currently has two built-in types:</p>
<ulclass="simple">
<li>A key-value store <codeclass="docutils literal"><spanclass="pre">QueryableStoreTypes#keyValueStore()</span></code>, see <aclass="reference internal"href="#streams-developer-guide-interactive-queries-local-key-value-stores"><spanclass="std std-ref">Querying local key-value stores</span></a>.</li>
<li>A window store <codeclass="docutils literal"><spanclass="pre">QueryableStoreTypes#windowStore()</span></code>, see <aclass="reference internal"href="#streams-developer-guide-interactive-queries-local-window-stores"><spanclass="std std-ref">Querying local window stores</span></a>.</li>
</ul>
<p>You can also <aclass="reference internal"href="#streams-developer-guide-interactive-queries-custom-stores"><spanclass="std std-ref">implement your own QueryableStoreType</span></a> as described in section <aclass="reference internal"href="#streams-developer-guide-interactive-queries-custom-stores"><spanclass="std std-ref">Querying local custom state stores</span></a>.</p>
<spanid="streams-developer-guide-interactive-queries-local-key-value-stores"></span><h3><aclass="toc-backref"href="#id4">Querying local key-value stores</a><aclass="headerlink"href="#querying-local-key-value-stores"title="Permalink to this headline"></a></h3>
<p>To query a local key-value store, you must first create a topology with a key-value store. This example creates a key-value
store named “CountsKeyValueStore”. This store will hold the latest count for any word that is found on the topic “word-count-input”.</p>
<p>After the application has started, you can get access to “CountsKeyValueStore” and then query it via the <aclass="reference external"href="https://github.com/apache/kafka/blob/1.0/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java">ReadOnlyKeyValueStore</a> API:</p>
<divclass="highlight-java"><divclass="highlight"><pre><span></span><spanclass="c1">// Get the key-value store CountsKeyValueStore</span>
<spanclass="n">System</span><spanclass="o">.</span><spanclass="na">out</span><spanclass="o">.</span><spanclass="na">println</span><spanclass="o">(</span><spanclass="s">"count for hello:"</span><spanclass="o">+</span><spanclass="n">keyValueStore</span><spanclass="o">.</span><spanclass="na">get</span><spanclass="o">(</span><spanclass="s">"hello"</span><spanclass="o">));</span>
<spanclass="c1">// Get the values for a range of keys available in this application instance</span>
<spanclass="n">System</span><spanclass="o">.</span><spanclass="na">out</span><spanclass="o">.</span><spanclass="na">println</span><spanclass="o">(</span><spanclass="s">"count for "</span><spanclass="o">+</span><spanclass="n">next</span><spanclass="o">.</span><spanclass="na">key</span><spanclass="o">+</span><spanclass="s">": "</span><spanclass="o">+</span><spanclass="n">value</span><spanclass="o">);</span>
<spanclass="o">}</span>
<spanclass="c1">// Get the values for all of the keys available in this application instance</span>
<spanclass="n">System</span><spanclass="o">.</span><spanclass="na">out</span><spanclass="o">.</span><spanclass="na">println</span><spanclass="o">(</span><spanclass="s">"count for "</span><spanclass="o">+</span><spanclass="n">next</span><spanclass="o">.</span><spanclass="na">key</span><spanclass="o">+</span><spanclass="s">": "</span><spanclass="o">+</span><spanclass="n">value</span><spanclass="o">);</span>
<spanclass="o">}</span>
</pre></div>
</div>
<p>You can also materialize the results of stateless operators by using the overloaded methods that take a <codeclass="docutils literal"><spanclass="pre">queryableStoreName</span></code>
<spanid="streams-developer-guide-interactive-queries-local-window-stores"></span><h3><aclass="toc-backref"href="#id5">Querying local window stores</a><aclass="headerlink"href="#querying-local-window-stores"title="Permalink to this headline"></a></h3>
<p>A window store will potentially have many results for any given key because the key can be present in multiple windows.
However, there is only one result per window for a given key.</p>
<p>To query a local window store, you must first create a topology with a window store. This example creates a window store
named “CountsWindowStore” that contains the counts for words in 1-minute windows.</p>
<p>After the application has started, you can get access to “CountsWindowStore” and then query it via the <aclass="reference external"href="https://github.com/apache/kafka/blob/1.0/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java">ReadOnlyWindowStore</a> API:</p>
<divclass="highlight-java"><divclass="highlight"><pre><span></span><spanclass="c1">// Get the window store named "CountsWindowStore"</span>
<spanclass="c1">// Fetch values for the key "world" for all of the windows available in this application instance.</span>
<spanclass="c1">// To get *all* available windows we fetch windows from the beginning of time until now.</span>
<spanclass="kt">long</span><spanclass="n">timeFrom</span><spanclass="o">=</span><spanclass="mi">0</span><spanclass="o">;</span><spanclass="c1">// beginning of time = oldest available</span>
<spanclass="kt">long</span><spanclass="n">timeTo</span><spanclass="o">=</span><spanclass="n">System</span><spanclass="o">.</span><spanclass="na">currentTimeMillis</span><spanclass="o">();</span><spanclass="c1">// now (in processing-time)</span>
<spanclass="n">System</span><spanclass="o">.</span><spanclass="na">out</span><spanclass="o">.</span><spanclass="na">println</span><spanclass="o">(</span><spanclass="s">"Count of 'world' @ time "</span><spanclass="o">+</span><spanclass="n">windowTimestamp</span><spanclass="o">+</span><spanclass="s">" is "</span><spanclass="o">+</span><spanclass="n">next</span><spanclass="o">.</span><spanclass="na">value</span><spanclass="o">);</span>
<spanid="streams-developer-guide-interactive-queries-custom-stores"></span><h3><aclass="toc-backref"href="#id6">Querying local custom state stores</a><aclass="headerlink"href="#querying-local-custom-state-stores"title="Permalink to this headline"></a></h3>
<pclass="last">Only the <aclass="reference internal"href="processor-api.html#streams-developer-guide-processor-api"><spanclass="std std-ref">Processor API</span></a> supports custom state stores.</p>
</div>
<p>Before querying the custom state stores you must implement these interfaces:</p>
<ulclass="simple">
<li>Your custom state store must implement <codeclass="docutils literal"><spanclass="pre">StateStore</span></code>.</li>
<li>You must have an interface to represent the operations available on the store.</li>
<li>You must provide an implementation of <codeclass="docutils literal"><spanclass="pre">StoreBuilder</span></code> for creating instances of your store.</li>
<li>It is recommended that you provide an interface that restricts access to read-only operations. This prevents users of this API from mutating the state of your running Kafka Streams application out-of-band.</li>
</ul>
<p>The class/interface hierarchy for your custom store might look something like:</p>
<spanclass="c1">// implementation of the supplier for MyCustomStore</span>
<spanclass="o">}</span>
</pre></div>
</div>
<p>To make this store queryable you must:</p>
<ulclass="simple">
<li>Provide an implementation of <aclass="reference external"href="https://github.com/apache/kafka/blob/1.0/streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java">QueryableStoreType</a>.</li>
<li>Provide a wrapper class that has access to all of the underlying instances of the store and is used for querying.</li>
</ul>
<p>Here is how to implement <codeclass="docutils literal"><spanclass="pre">QueryableStoreType</span></code>:</p>
interface to get access to the underlying instances of your store.
<codeclass="docutils literal"><spanclass="pre">StateStoreProvider#stores(String</span><spanclass="pre">storeName,</span><spanclass="pre">QueryableStoreType<T></span><spanclass="pre">queryableStoreType)</span></code> returns a <codeclass="docutils literal"><spanclass="pre">List</span></code> of state
stores with the given storeName and of the type as defined by <codeclass="docutils literal"><spanclass="pre">queryableStoreType</span></code>.</p>
<p>Here is an example implementation of the wrapper follows (Java 8+):</p>
<divclass="highlight-java"><divclass="highlight"><pre><span></span><spanclass="c1">// We strongly recommended implementing a read-only interface</span>
<spanclass="c1">// to restrict usage of the store to safe read operations!</span>
<spanid="streams-developer-guide-interactive-queries-discovery"></span><h2><aclass="toc-backref"href="#id7">Querying remote state stores for the entire app</a><aclass="headerlink"href="#querying-remote-state-stores-for-the-entire-app"title="Permalink to this headline"></a></h2>
<p>To query remote states for the entire app, you must expose the application’s full state to other applications, including
applications that are running on different machines.</p>
<p>For example, you have a Kafka Streams application that processes user events in a multi-player video game, and you want to retrieve the latest status of each user directly and display it in a mobile app. Here are the required steps to make the full state of your application queryable:</p>
<olclass="arabic simple">
<li><aclass="reference internal"href="#streams-developer-guide-interactive-queries-rpc-layer"><spanclass="std std-ref">Add an RPC layer to your application</span></a> so that
the instances of your application can be interacted with via the network (e.g., a REST API, Thrift, a custom protocol,
and so on). The instances must respond to interactive queries. You can follow the reference examples provided to get
started.</li>
<li><aclass="reference internal"href="#streams-developer-guide-interactive-queries-expose-rpc"><spanclass="std std-ref">Expose the RPC endpoints</span></a> of
your application’s instances via the <codeclass="docutils literal"><spanclass="pre">application.server</span></code> configuration setting of Kafka Streams. Because RPC
endpoints must be unique within a network, each instance has its own value for this configuration setting.
This makes an application instance discoverable by other instances.</li>
<li>In the RPC layer, <aclass="reference internal"href="#streams-developer-guide-interactive-queries-discover-app-instances-and-stores"><spanclass="std std-ref">discover remote application instances</span></a> and their state stores and <aclass="reference internal"href="#streams-developer-guide-interactive-queries-local-stores"><spanclass="std std-ref">query locally available state stores</span></a> to make the full state of your application queryable. The remote application instances can forward queries to other app instances if a particular instance lacks the local data to respond to a query. The locally available state stores can directly respond to queries.</li>
<pclass="caption"><spanclass="caption-text">Discover any running instances of the same application as well as the respective RPC endpoints they expose for
<spanid="streams-developer-guide-interactive-queries-rpc-layer"></span><h3><aclass="toc-backref"href="#id8">Adding an RPC layer to your application</a><aclass="headerlink"href="#adding-an-rpc-layer-to-your-application"title="Permalink to this headline"></a></h3>
<p>There are many ways to add an RPC layer. The only requirements are that the RPC layer is embedded within the Kafka Streams
application and that it exposes an endpoint that other application instances and applications can connect to.</p>
<spanid="streams-developer-guide-interactive-queries-expose-rpc"></span><h3><aclass="toc-backref"href="#id9">Exposing the RPC endpoints of your application</a><aclass="headerlink"href="#exposing-the-rpc-endpoints-of-your-application"title="Permalink to this headline"></a></h3>
<p>To enable remote state store discovery in a distributed Kafka Streams application, you must set the <aclass="reference internal"href="config-streams.html#streams-developer-guide-required-configs"><spanclass="std std-ref">configuration property</span></a> in the config properties.
The <codeclass="docutils literal"><spanclass="pre">application.server</span></code> property defines a unique <codeclass="docutils literal"><spanclass="pre">host:port</span></code> pair that points to the RPC endpoint of the respective instance of a Kafka Streams application.
The value of this configuration property will vary across the instances of your application.
When this property is set, Kafka Streams will keep track of the RPC endpoint information for every instance of an application, its state stores, and assigned stream partitions through instances of <aclass="reference external"href="../../../javadoc/org/apache/kafka/streams/state/StreamsMetadata.html">StreamsMetadata</a>.</p>
<spanid="streams-developer-guide-interactive-queries-discover-app-instances-and-stores"></span><h3><aclass="toc-backref"href="#id10">Discovering and accessing application instances and their local state stores</a><aclass="headerlink"href="#discovering-and-accessing-application-instances-and-their-local-state-stores"title="Permalink to this headline"></a></h3>
<p>The following methods return <aclass="reference external"href="../../../javadoc/org/apache/kafka/streams/state/StreamsMetadata.html">StreamsMetadata</a> objects, which provide meta-information about application instances such as their RPC endpoint and locally available state stores.</p>
<li><codeclass="docutils literal"><spanclass="pre">KafkaStreams#allMetadata()</span></code>: find all instances of this application</li>
<li><codeclass="docutils literal"><spanclass="pre">KafkaStreams#allMetadataForStore(String</span><spanclass="pre">storeName)</span></code>: find those applications instances that manage local instances of the state store “storeName”</li>
<li><codeclass="docutils literal"><spanclass="pre">KafkaStreams#metadataForKey(String</span><spanclass="pre">storeName,</span><spanclass="pre">K</span><spanclass="pre">key,</span><spanclass="pre">Serializer<K></span><spanclass="pre">keySerializer)</span></code>: using the default stream partitioning strategy, find the one application instance that holds the data for the given key in the given state store</li>
<li><codeclass="docutils literal"><spanclass="pre">KafkaStreams#metadataForKey(String</span><spanclass="pre">storeName,</span><spanclass="pre">K</span><spanclass="pre">key,</span><spanclass="pre">StreamPartitioner<K,</span><spanclass="pre">?></span><spanclass="pre">partitioner)</span></code>: using <codeclass="docutils literal"><spanclass="pre">partitioner</span></code>, find the one application instance that holds the data for the given key in the given state store</li>
<pclass="last">If <codeclass="docutils literal"><spanclass="pre">application.server</span></code> is not configured for an application instance, then the above methods will not find any <aclass="reference external"href="../../../javadoc/org/apache/kafka/streams/state/StreamsMetadata.html">StreamsMetadata</a> for it.</p>
<p>For example, we can now find the <codeclass="docutils literal"><spanclass="pre">StreamsMetadata</span></code> for the state store named “word-count” that we defined in the
