Class KafkaStreamBrancher<K,V>

java.lang.Object
org.springframework.kafka.support.KafkaStreamBrancher<K,V>
Type Parameters:
K - Type of keys
V - Type of values
public final class KafkaStreamBrancher<K,V> extends Object
Provides a method-chaining way to build branches in Kafka Streams processor topology.

Example of usage: 

 
 new KafkaStreamBrancher<String, String>()
    .branch((key, value) -> value.contains("A"), ks->ks.to("A"))
    .branch((key, value) -> value.contains("B"), ks->ks.to("B"))
    //default branch should not necessarily be defined in the end
    .defaultBranch(ks->ks.to("C"))
    .onTopOf(builder.stream("source"))
Since:
2.2.4
Author:
Ivan Ponomarev, Artem Bilan, Soby Chacko

  • Constructor Details

    • KafkaStreamBrancher

      public KafkaStreamBrancher()

  • Method Details

    • branch

      public KafkaStreamBrancher<K,V> branch(org.apache.kafka.streams.kstream.Predicate<? super K,? super V> predicate, Consumer<? super org.apache.kafka.streams.kstream.KStream<K,V>> consumer)
      Defines a new branch.
      Parameters:
      predicate - Predicate instance
      consumer - The consumer of this branch's KStream
      Returns:
      this

    • defaultBranch

      public KafkaStreamBrancher<K,V> defaultBranch(Consumer<? super org.apache.kafka.streams.kstream.KStream<K,V>> consumer)
      Defines a default branch. All the messages that were not dispatched to other branches will be directed to this stream. This method should not necessarily be called in the end of chain.
      Parameters:
      consumer - The consumer of this branch's KStream
      Returns:
      this

    • onTopOf

      public org.apache.kafka.streams.kstream.KStream<K,V> onTopOf(org.apache.kafka.streams.kstream.KStream<K,V> stream)
      Terminating method that builds branches on top of given KStream. Applies each predicate-consumer pair sequentially to create branches. If a default consumer exists, it will handle all records that don't match any predicates.
      Parameters:
      stream - KStream to split
      Returns:
      the processed stream
      Throws:
      NullPointerException - if stream is null
      IllegalStateException - if number of predicates doesn't match number of consumers