Package org.springframework.cloud.stream.binder.kafka.streams.annotations

Annotation Type KafkaStreamsStateStore

@Target({TYPE,METHOD,ANNOTATION_TYPE}) @Retention(RUNTIME) public @interface KafkaStreamsStateStore
Interface for Kafka Stream state store. This interface can be used to inject a state store specification into KStream building process so that the desired store can be built by StreamBuilder and added to topology for later use by processors. This is particularly useful when need to combine stream DSL with low level processor APIs. In those cases, if a writable state store is desired in processors, it needs to be created using this annotation. Here is the example. 
     @StreamListener("input")
     @KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW,
                                                                size=300000)
           public void process(KStream<Object, Product> input) {
         ......
     }
With that, you should be able to read/write this state store in your processor/transformer code. 
                new Processor<Object, Product>() {
                        WindowStore<Object, String> state;
                        @Override
                        public void init(ProcessorContext processorContext) {
                        state = (WindowStore)processorContext.getStateStore("mystate");
                                ......
                        }
                }
Author:
Lei Chen

  • Element Details

    • name

      String name
      Provides name of the state store.
      Returns:
      name of state store.
      Default:
      ""

    • type

      KafkaStreamsStateStoreProperties.StoreType type
      State store type.
      Returns:
      KafkaStreamsStateStoreProperties.StoreType of state store.
      Default:
      KEYVALUE

    • keySerde

      String keySerde
      Serde used for key.
      Returns:
      key serde of state store.
      Default:
      "org.apache.kafka.common.serialization.Serdes$StringSerde"

    • valueSerde

      String valueSerde
      Serde used for value.
      Returns:
      value serde of state store.
      Default:
      "org.apache.kafka.common.serialization.Serdes$StringSerde"

    • lengthMs

      long lengthMs
      Length in milli-second of Windowed store window.
      Returns:
      length in milli-second of window(for windowed store).
      Default:
      0L

    • retentionMs

      long retentionMs
      Retention period for Windowed store windows.
      Returns:
      the maximum period of time in milli-second to keep each window in this store(for windowed store).
      Default:
      0L

    • cache

      boolean cache
      Whether catching is enabled or not.
      Returns:
      whether caching should be enabled on the created store.
      Default:
      false

    • logging

      boolean logging
      Whether logging is enabled or not.
      Returns:
      whether logging should be enabled on the created store.
      Default:
      true