Field Detail

• internalTopologyBuilder

  protected final org.apache.kafka.streams.processor.internals.InternalTopologyBuilder internalTopologyBuilder

Constructor Detail

• Topology

  public Topology()

• Topology

  public Topology(TopologyConfig topologyConfigs)

• Topology

  protected Topology(org.apache.kafka.streams.processor.internals.InternalTopologyBuilder internalTopologyBuilder)

Method Detail

• addSource

  public Topology addSource(String name,
                            String... topics)

  Add a new source that consumes the named topics and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration. The default TimestampExtractor as specified in the config is used.

  Parameters:

  name - the unique name of the source used to reference this node when adding processor children.

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(String name,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration. The default TimestampExtractor as specified in the config is used.

  Parameters:

  name - the unique name of the source used to reference this node when adding processor children.

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            String... topics)

  Add a new source that consumes the named topics and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration. The default TimestampExtractor as specified in the config is used.

  Parameters:

  offsetReset - the auto offset reset policy to use for this source if no committed offsets found; acceptable values earliest or latest

  name - the unique name of the source used to reference this node when adding processor children.

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration. The default TimestampExtractor as specified in the config is used.

  Parameters:

  offsetReset - the auto offset reset policy value for this source if no committed offsets found; acceptable values earliest or latest.

  name - the unique name of the source used to reference this node when adding processor children.

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(TimestampExtractor timestampExtractor,
                            String name,
                            String... topics)

  Add a new source that consumes the named topics and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration.

  Parameters:

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  name - the unique name of the source used to reference this node when adding processor children.

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(TimestampExtractor timestampExtractor,
                            String name,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration.

  Parameters:

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  name - the unique name of the source used to reference this node when adding processor children.

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            TimestampExtractor timestampExtractor,
                            String name,
                            String... topics)

  Add a new source that consumes the named topics and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration.

  Parameters:

  offsetReset - the auto offset reset policy to use for this source if no committed offsets found; acceptable values earliest or latest

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  name - the unique name of the source used to reference this node when adding processor children.

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            TimestampExtractor timestampExtractor,
                            String name,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forward the records to child processor and/or sink nodes. The source will use the default key deserializer and default value deserializer specified in the stream configuration.

  Parameters:

  offsetReset - the auto offset reset policy value for this source if no committed offsets found; acceptable values earliest or latest.

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  name - the unique name of the source used to reference this node when adding processor children.

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(String name,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            String... topics)

  Add a new source that consumes the named topics and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers. The default TimestampExtractor as specified in the config is used.

  Parameters:

  name - the unique name of the source used to reference this node when adding processor children

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(String name,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers. The provided de-/serializers will be used for all matched topics, so care should be taken to specify patterns for topics that share the same key-value data format. The default TimestampExtractor as specified in the config is used.

  Parameters:

  name - the unique name of the source used to reference this node when adding processor children

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by name

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            String... topics)

  Add a new source that consumes from topics matching the given pattern and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers. The provided de-/serializers will be used for all the specified topics, so care should be taken when specifying topics that share the same key-value data format.

  Parameters:

  offsetReset - the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest

  name - the unique name of the source used to reference this node when adding processor children

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by name

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers. The provided de-/serializers will be used for all matched topics, so care should be taken to specify patterns for topics that share the same key-value data format.

  Parameters:

  offsetReset - the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest

  name - the unique name of the source used to reference this node when adding processor children

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by name

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            TimestampExtractor timestampExtractor,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            String... topics)

  Add a new source that consumes the named topics and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers.

  Parameters:

  offsetReset - the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest.

  name - the unique name of the source used to reference this node when adding processor children.

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topics - the name of one or more Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by another source

• addSource

  public Topology addSource(Topology.AutoOffsetReset offsetReset,
                            String name,
                            TimestampExtractor timestampExtractor,
                            org.apache.kafka.common.serialization.Deserializer<?> keyDeserializer,
                            org.apache.kafka.common.serialization.Deserializer<?> valueDeserializer,
                            Pattern topicPattern)

  Add a new source that consumes from topics matching the given pattern and forwards the records to child processor and/or sink nodes. The source will use the specified key and value deserializers. The provided de-/serializers will be used for all matched topics, so care should be taken to specify patterns for topics that share the same key-value data format.

  Parameters:

  offsetReset - the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest

  name - the unique name of the source used to reference this node when adding processor children.

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  keyDeserializer - key deserializer used to read this source, if not specified the default key deserializer defined in the configs will be used

  valueDeserializer - value deserializer used to read this source, if not specified the default value deserializer defined in the configs will be used

  topicPattern - regular expression pattern to match Kafka topics that this source is to consume

  Returns:

  itself

  Throws:

  TopologyException - if processor is already added or if topics have already been registered by name

• addSink

  public Topology addSink(String name,
                          String topic,
                          String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to the named Kafka topic. The sink will use the default key serializer and default value serializer specified in the stream configuration.

  Parameters:

  name - the unique name of the sink

  topic - the name of the Kafka topic to which this sink should write its records

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and write to its topic

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                String topic,
                                StreamPartitioner<? super K,? super V> partitioner,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to the named Kafka topic, using the supplied partitioner. The sink will use the default key serializer and default value serializer specified in the stream configuration.

  The sink will also use the specified StreamPartitioner to determine how records are distributed among the named Kafka topic's partitions. Such control is often useful with topologies that use state stores in its processors. In most other cases, however, a partitioner needs not be specified and Kafka will automatically distribute records among partitions using Kafka's default partitioning logic.

  Parameters:

  name - the unique name of the sink

  topic - the name of the Kafka topic to which this sink should write its records

  partitioner - the function that should be used to determine the partition for each record processed by the sink

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and write to its topic

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, Serializer, Serializer, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                String topic,
                                org.apache.kafka.common.serialization.Serializer<K> keySerializer,
                                org.apache.kafka.common.serialization.Serializer<V> valueSerializer,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to the named Kafka topic. The sink will use the specified key and value serializers.

  Parameters:

  name - the unique name of the sink

  topic - the name of the Kafka topic to which this sink should write its records

  keySerializer - the key serializer used when consuming records; may be null if the sink should use the default key serializer specified in the stream configuration

  valueSerializer - the value serializer used when consuming records; may be null if the sink should use the default value serializer specified in the stream configuration

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and write to its topic

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                String topic,
                                org.apache.kafka.common.serialization.Serializer<K> keySerializer,
                                org.apache.kafka.common.serialization.Serializer<V> valueSerializer,
                                StreamPartitioner<? super K,? super V> partitioner,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to the named Kafka topic. The sink will use the specified key and value serializers, and the supplied partitioner.

  Parameters:

  name - the unique name of the sink

  topic - the name of the Kafka topic to which this sink should write its records

  keySerializer - the key serializer used when consuming records; may be null if the sink should use the default key serializer specified in the stream configuration

  valueSerializer - the value serializer used when consuming records; may be null if the sink should use the default value serializer specified in the stream configuration

  partitioner - the function that should be used to determine the partition for each record processed by the sink

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and write to its topic

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                TopicNameExtractor<K,V> topicExtractor,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to Kafka topics based on topicExtractor. The topics that it may ever send to should be pre-created. The sink will use the default key serializer and default value serializer specified in the stream configuration.

  Parameters:

  name - the unique name of the sink

  topicExtractor - the extractor to determine the name of the Kafka topic to which this sink should write for each record

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and dynamically write to topics

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                TopicNameExtractor<K,V> topicExtractor,
                                StreamPartitioner<? super K,? super V> partitioner,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to Kafka topics based on topicExtractor, using the supplied partitioner. The topics that it may ever send to should be pre-created. The sink will use the default key serializer and default value serializer specified in the stream configuration.

  The sink will also use the specified StreamPartitioner to determine how records are distributed among the named Kafka topic's partitions. Such control is often useful with topologies that use state stores in its processors. In most other cases, however, a partitioner needs not be specified and Kafka will automatically distribute records among partitions using Kafka's default partitioning logic.

  Parameters:

  name - the unique name of the sink

  topicExtractor - the extractor to determine the name of the Kafka topic to which this sink should write for each record

  partitioner - the function that should be used to determine the partition for each record processed by the sink

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and dynamically write to topics

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, Serializer, Serializer, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                TopicNameExtractor<K,V> topicExtractor,
                                org.apache.kafka.common.serialization.Serializer<K> keySerializer,
                                org.apache.kafka.common.serialization.Serializer<V> valueSerializer,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to Kafka topics based on topicExtractor. The topics that it may ever send to should be pre-created. The sink will use the specified key and value serializers.

  Parameters:

  name - the unique name of the sink

  topicExtractor - the extractor to determine the name of the Kafka topic to which this sink should write for each record

  keySerializer - the key serializer used when consuming records; may be null if the sink should use the default key serializer specified in the stream configuration

  valueSerializer - the value serializer used when consuming records; may be null if the sink should use the default value serializer specified in the stream configuration

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and dynamically write to topics

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)

• addSink

  public <K,V> Topology addSink(String name,
                                TopicNameExtractor<K,V> topicExtractor,
                                org.apache.kafka.common.serialization.Serializer<K> keySerializer,
                                org.apache.kafka.common.serialization.Serializer<V> valueSerializer,
                                StreamPartitioner<? super K,? super V> partitioner,
                                String... parentNames)

  Add a new sink that forwards records from upstream parent processor and/or source nodes to Kafka topics based on topicExtractor. The topics that it may ever send to should be pre-created. The sink will use the specified key and value serializers, and the supplied partitioner.

  Parameters:

  name - the unique name of the sink

  topicExtractor - the extractor to determine the name of the Kafka topic to which this sink should write for each record

  keySerializer - the key serializer used when consuming records; may be null if the sink should use the default key serializer specified in the stream configuration

  valueSerializer - the value serializer used when consuming records; may be null if the sink should use the default value serializer specified in the stream configuration

  partitioner - the function that should be used to determine the partition for each record processed by the sink

  parentNames - the name of one or more source or processor nodes whose output records this sink should consume and dynamically write to topics

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

  See Also:

  addSink(String, String, String...), addSink(String, String, StreamPartitioner, String...), addSink(String, String, Serializer, Serializer, String...)

• addProcessor

  @Deprecated
  public Topology addProcessor(String name,
                                           ProcessorSupplier supplier,
                                           String... parentNames)

  Deprecated. Since 2.7.0 Use addProcessor(String, ProcessorSupplier, String...) instead.
  Add a new processor node that receives and processes records output by one or more parent source or processor node. Any new record output by this processor will be forwarded to its child processor or sink nodes. The supplier should always generate a new instance each time ProcessorSupplier.get() gets called. Creating a single Processor object and returning the same object reference in ProcessorSupplier.get() would be a violation of the supplier pattern and leads to runtime exceptions. If supplier provides stores via ConnectedStoreProvider.stores(), the provided StoreBuilders will be added to the topology and connected to this processor automatically.

  Parameters:

  name - the unique name of the processor node

  supplier - the supplier used to obtain this node's Processor instance

  parentNames - the name of one or more source or processor nodes whose output records this processor should receive and process

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

• addProcessor

  public <KIn,VIn,KOut,VOut> Topology addProcessor(String name,
                                                   ProcessorSupplier<KIn,VIn,KOut,VOut> supplier,
                                                   String... parentNames)

  Add a new processor node that receives and processes records output by one or more parent source or processor node. Any new record output by this processor will be forwarded to its child processor or sink nodes. If supplier provides stores via ConnectedStoreProvider.stores(), the provided StoreBuilders will be added to the topology and connected to this processor automatically.

  Parameters:

  name - the unique name of the processor node

  supplier - the supplier used to obtain this node's Processor instance

  parentNames - the name of one or more source or processor nodes whose output records this processor should receive and process

  Returns:

  itself

  Throws:

  TopologyException - if parent processor is not added yet, or if this processor's name is equal to the parent's name

• addStateStore

  public Topology addStateStore(StoreBuilder<?> storeBuilder,
                                String... processorNames)

  Adds a state store.

  Parameters:

  storeBuilder - the storeBuilder used to obtain this state store StateStore instance

  processorNames - the names of the processors that should be able to access the provided store

  Returns:

  itself

  Throws:

  TopologyException - if state store supplier is already added

• addGlobalStore

  @Deprecated
  public <K,V> Topology addGlobalStore(StoreBuilder<?> storeBuilder,
                                                   String sourceName,
                                                   org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                   org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer,
                                                   String topic,
                                                   String processorName,
                                                   ProcessorSupplier<K,V> stateUpdateSupplier)

  Deprecated. Since 2.7.0. Use addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier) instead.
  Adds a global StateStore to the topology. The StateStore sources its data from all partitions of the provided input topic. There will be exactly one instance of this StateStore per Kafka Streams instance.

  A SourceNode with the provided sourceName will be added to consume the data arriving from the partitions of the input topic.

  The provided ProcessorSupplier will be used to create an ProcessorNode that will receive all records forwarded from the SourceNode. The supplier should always generate a new instance each time ProcessorSupplier.get() gets called. Creating a single Processor object and returning the same object reference in ProcessorSupplier.get() would be a violation of the supplier pattern and leads to runtime exceptions. This ProcessorNode should be used to keep the StateStore up-to-date. The default TimestampExtractor as specified in the config is used.

  Parameters:

  storeBuilder - user defined state store builder

  sourceName - name of the SourceNode that will be automatically added

  keyDeserializer - the Deserializer to deserialize keys with

  valueDeserializer - the Deserializer to deserialize values with

  topic - the topic to source the data from

  processorName - the name of the ProcessorSupplier

  stateUpdateSupplier - the instance of ProcessorSupplier

  Returns:

  itself

  Throws:

  TopologyException - if the processor of state is already registered

• addGlobalStore

  @Deprecated
  public <K,V> Topology addGlobalStore(StoreBuilder<?> storeBuilder,
                                                   String sourceName,
                                                   TimestampExtractor timestampExtractor,
                                                   org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer,
                                                   org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer,
                                                   String topic,
                                                   String processorName,
                                                   ProcessorSupplier<K,V> stateUpdateSupplier)

  Deprecated. Since 2.7.0. Use addGlobalStore(StoreBuilder, String, TimestampExtractor, Deserializer, Deserializer, String, String, ProcessorSupplier) instead.
  Adds a global StateStore to the topology. The StateStore sources its data from all partitions of the provided input topic. There will be exactly one instance of this StateStore per Kafka Streams instance.

  A SourceNode with the provided sourceName will be added to consume the data arriving from the partitions of the input topic.

  The provided ProcessorSupplier will be used to create an ProcessorNode that will receive all records forwarded from the SourceNode. The supplier should always generate a new instance each time ProcessorSupplier.get() gets called. Creating a single Processor object and returning the same object reference in ProcessorSupplier.get() would be a violation of the supplier pattern and leads to runtime exceptions. This ProcessorNode should be used to keep the StateStore up-to-date.

  Parameters:

  storeBuilder - user defined key value store builder

  sourceName - name of the SourceNode that will be automatically added

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  keyDeserializer - the Deserializer to deserialize keys with

  valueDeserializer - the Deserializer to deserialize values with

  topic - the topic to source the data from

  processorName - the name of the ProcessorSupplier

  stateUpdateSupplier - the instance of ProcessorSupplier

  Returns:

  itself

  Throws:

  TopologyException - if the processor of state is already registered

• addGlobalStore

  public <KIn,VIn> Topology addGlobalStore(StoreBuilder<?> storeBuilder,
                                           String sourceName,
                                           org.apache.kafka.common.serialization.Deserializer<KIn> keyDeserializer,
                                           org.apache.kafka.common.serialization.Deserializer<VIn> valueDeserializer,
                                           String topic,
                                           String processorName,
                                           ProcessorSupplier<KIn,VIn,Void,Void> stateUpdateSupplier)

  Adds a global StateStore to the topology. The StateStore sources its data from all partitions of the provided input topic. There will be exactly one instance of this StateStore per Kafka Streams instance.

  A SourceNode with the provided sourceName will be added to consume the data arriving from the partitions of the input topic.

  The provided ProcessorSupplier will be used to create an ProcessorNode that will receive all records forwarded from the SourceNode. This ProcessorNode should be used to keep the StateStore up-to-date. The default TimestampExtractor as specified in the config is used.

  Parameters:

  storeBuilder - user defined state store builder

  sourceName - name of the SourceNode that will be automatically added

  keyDeserializer - the Deserializer to deserialize keys with

  valueDeserializer - the Deserializer to deserialize values with

  topic - the topic to source the data from

  processorName - the name of the ProcessorSupplier

  stateUpdateSupplier - the instance of ProcessorSupplier

  Returns:

  itself

  Throws:

  TopologyException - if the processor of state is already registered

• addGlobalStore

  public <KIn,VIn> Topology addGlobalStore(StoreBuilder<?> storeBuilder,
                                           String sourceName,
                                           TimestampExtractor timestampExtractor,
                                           org.apache.kafka.common.serialization.Deserializer<KIn> keyDeserializer,
                                           org.apache.kafka.common.serialization.Deserializer<VIn> valueDeserializer,
                                           String topic,
                                           String processorName,
                                           ProcessorSupplier<KIn,VIn,Void,Void> stateUpdateSupplier)

  Adds a global StateStore to the topology. The StateStore sources its data from all partitions of the provided input topic. There will be exactly one instance of this StateStore per Kafka Streams instance.

  A SourceNode with the provided sourceName will be added to consume the data arriving from the partitions of the input topic.

  The provided ProcessorSupplier will be used to create an ProcessorNode that will receive all records forwarded from the SourceNode. This ProcessorNode should be used to keep the StateStore up-to-date.

  Parameters:

  storeBuilder - user defined key value store builder

  sourceName - name of the SourceNode that will be automatically added

  timestampExtractor - the stateless timestamp extractor used for this source, if not specified the default extractor defined in the configs will be used

  keyDeserializer - the Deserializer to deserialize keys with

  valueDeserializer - the Deserializer to deserialize values with

  topic - the topic to source the data from

  processorName - the name of the ProcessorSupplier

  stateUpdateSupplier - the instance of ProcessorSupplier

  Returns:

  itself

  Throws:

  TopologyException - if the processor of state is already registered

• connectProcessorAndStateStores

  public Topology connectProcessorAndStateStores(String processorName,
                                                 String... stateStoreNames)

  Connects the processor and the state stores.

  Parameters:

  processorName - the name of the processor

  stateStoreNames - the names of state stores that the processor uses

  Returns:

  itself

  Throws:

  TopologyException - if the processor or a state store is unknown

• describe

  public TopologyDescription describe()

  Returns a description of the specified Topology.

  Returns:

  a description of the topology.