Polishing Kafka Streams binder docs

Resolves #390
This commit is contained in:
Soby Chacko
2018-06-27 11:25:22 -04:00
parent 54d7c333d3
commit b09def9ccc

View File

@@ -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
}
----