Package org.apache.kafka.streams

Class Topology

java.lang.Object

org.apache.kafka.streams.Topology

public class Topology
extends Object

A logical representation of a ProcessorTopology. A topology is a graph of sources, processors, and sinks. A SourceNode is a node in the graph that consumes one or more Kafka topics and forwards them to its successor nodes. A Processor is a node in the graph that receives input records from upstream nodes, processes the records, and optionally forwarding new records to one, multiple, or all of its downstream nodes. Finally, a SinkNode is a node in the graph that receives records from upstream nodes and writes them to a Kafka topic. A Topology allows you to construct a graph of these nodes, and then passed into a new KafkaStreams instance that will then begin consuming, processing, and producing records.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static enum	Topology.AutoOffsetReset	Deprecated. Since 4.0.

Constructor Summary

Constructors

Constructor	Description
Topology()
Topology(TopologyConfig topologyConfigs)

Method Summary

Modifier and Type	Method	Description
<K, V, S extends StateStore> Topology	addGlobalStore(StoreBuilder<S> storeBuilder, String sourceName, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String topic, String processorName, ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)	Adds a global state store to the topology.
<K, V, S extends StateStore> Topology	addGlobalStore(StoreBuilder<S> storeBuilder, String sourceName, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String topic, String processorName, ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)	See addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier).
<KIn, VIn, KOut, VOut> Topology	addProcessor(String name, ProcessorSupplier<KIn,VIn,KOut,VOut> processorSupplier, String... parentNames)	Add a processor that receives and processed records from one or more parent processors or sources.
<K, V, S extends StateStore> Topology	addReadOnlyStateStore(StoreBuilder<S> storeBuilder, String sourceName, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String topic, String processorName, ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)	Adds a read-only state store to the topology.
<K, V, S extends StateStore> Topology	addReadOnlyStateStore(StoreBuilder<S> storeBuilder, String sourceName, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String topic, String processorName, ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)	See addReadOnlyStateStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier).
Topology	addSink(String name, String topic, String... parentNames)	Add a sink that sends records from upstream processors or sources to the named Kafka topic.
<K, V> Topology	addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valueSerializer, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, String topic, Serializer<K> keySerializer, Serializer<V> valueSerializer, StreamPartitioner<? super K,? super V> partitioner, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, String topic, StreamPartitioner<? super K,? super V> partitioner, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, TopicNameExtractor<? super K,? super V> topicExtractor, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, TopicNameExtractor<? super K,? super V> topicExtractor, Serializer<K> keySerializer, Serializer<V> valueSerializer, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, TopicNameExtractor<? super K,? super V> topicExtractor, Serializer<K> keySerializer, Serializer<V> valueSerializer, StreamPartitioner<? super K,? super V> partitioner, String... parentNames)	See addSink(String, String, String...).
<K, V> Topology	addSink(String name, TopicNameExtractor<? super K,? super V> topicExtractor, StreamPartitioner<? super K,? super V> partitioner, String... parentNames)	See addSink(String, String, String...).
Topology	addSource(String name, String... topics)	Add a source that consumes the named topics and forwards the records to child processors and sinks.
Topology	addSource(String name, Pattern topicPattern)	See addSource(String, String...).
<K, V> Topology	addSource(String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String... topics)	See addSource(String, String...).
<K, V> Topology	addSource(String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Pattern topicPattern)	See addSource(String, Pattern).
Topology	addSource(AutoOffsetReset offsetReset, String name, String... topics)	See addSource(String, String...).
Topology	addSource(AutoOffsetReset offsetReset, String name, Pattern topicPattern)	See addSource(String, Pattern).
<K, V> Topology	addSource(AutoOffsetReset offsetReset, String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String... topics)	See addSource(String, String...).
<K, V> Topology	addSource(AutoOffsetReset offsetReset, String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Pattern topicPattern)	See addSource(String, Pattern).
<K, V> Topology	addSource(AutoOffsetReset offsetReset, String name, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String... topics)	See addSource(String, String...).
<K, V> Topology	addSource(AutoOffsetReset offsetReset, String name, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Pattern topicPattern)	See addSource(String, Pattern).
Topology	addSource(AutoOffsetReset offsetReset, TimestampExtractor timestampExtractor, String name, String... topics)	See addSource(String, String...).
Topology	addSource(AutoOffsetReset offsetReset, TimestampExtractor timestampExtractor, String name, Pattern topicPattern)	See addSource(String, Pattern).
Topology	addSource(TimestampExtractor timestampExtractor, String name, String... topics)	See addSource(String, String...).
Topology	addSource(TimestampExtractor timestampExtractor, String name, Pattern topicPattern)	See addSource(String, Pattern).
Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, String... topics)	Deprecated. Since 4.0.
Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, Pattern topicPattern)	Deprecated. Since 4.0.
<K, V> Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String... topics)	Deprecated. Since 4.0.
<K, V> Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Pattern topicPattern)	Deprecated. Since 4.0.
<K, V> Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, String... topics)	Deprecated. Since 4.0.
<K, V> Topology	addSource(Topology.AutoOffsetReset offsetReset, String name, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Pattern topicPattern)	Deprecated. Since 4.0.
Topology	addSource(Topology.AutoOffsetReset offsetReset, TimestampExtractor timestampExtractor, String name, String... topics)	Deprecated. Since 4.0.
Topology	addSource(Topology.AutoOffsetReset offsetReset, TimestampExtractor timestampExtractor, String name, Pattern topicPattern)	Deprecated. Since 4.0.
<S extends StateStore> Topology	addStateStore(StoreBuilder<S> storeBuilder, String... processorNames)	Add a state store to the topology, and optionally connect it to one or more processors.
Topology	connectProcessorAndStateStores(String processorName, String... stateStoreNames)	Connect a processor to one or more state stores.
TopologyDescription	describe()	Returns a description of the specified Topology.

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

Topology

public Topology()

Topology

public Topology(TopologyConfig topologyConfigs)

Method Details

addSource

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

Add a source that consumes the named topics and forwards the records to child processors and sinks.

The source will use the default values from StreamsConfig for

• key deserializer
• value deserializer
• auto.offset.reset
• timestamp extractor

If you want to specify a source specific auto.offset.reset strategy, TimestampExtractor, or key/value Deserializer, use the corresponding overloaded addSource(...) method.

Parameters:

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

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

Returns:

itself

Throws:

TopologyException - if the provided source name is not unique, no topics are specified, or a topic has already been registered by another source, read-only state store, or global state store

NullPointerException - if name or topics is null, or topics contains a null topic

See Also:

• addSource(String, Pattern)

addSource

public Topology addSource(String name,
 Pattern topicPattern)

See addSource(String, String...).

Takes a Pattern (cannot be null) to match topics to consumes from, instead of a list of topic names.

addSource

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

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, String...) instead.

addSource

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

See addSource(String, String...).

addSource

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

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, Pattern) instead.

addSource

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

See addSource(String, Pattern).

addSource

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

See addSource(String, String...).

addSource

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

See addSource(String, Pattern).

addSource

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

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, TimestampExtractor, String, String...) instead.

addSource

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

See addSource(String, String...).

addSource

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

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, TimestampExtractor, String, Pattern) instead.

addSource

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

See addSource(String, Pattern).

addSource

public <K,
V> Topology addSource(String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String... topics)

See addSource(String, String...).

addSource

public <K,
V> Topology addSource(String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 Pattern topicPattern)

See addSource(String, Pattern).

addSource

@Deprecated
public <K,
V> Topology addSource(Topology.AutoOffsetReset offsetReset,
 String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String... topics)

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, Deserializer, Deserializer, String...) instead.

addSource

public <K,
V> Topology addSource(AutoOffsetReset offsetReset,
 String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String... topics)

See addSource(String, String...).

addSource

@Deprecated
public <K,
V> Topology addSource(Topology.AutoOffsetReset offsetReset,
 String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 Pattern topicPattern)

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, Deserializer, Deserializer, Pattern) instead.

addSource

public <K,
V> Topology addSource(AutoOffsetReset offsetReset,
 String name,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 Pattern topicPattern)

See addSource(String, Pattern).

addSource

@Deprecated
public <K,
V> Topology addSource(Topology.AutoOffsetReset offsetReset,
 String name,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String... topics)

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, TimestampExtractor, Deserializer, Deserializer, String...) instead.

addSource

public <K,
V> Topology addSource(AutoOffsetReset offsetReset,
 String name,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String... topics)

See addSource(String, String...).

addSource

@Deprecated
public <K,
V> Topology addSource(Topology.AutoOffsetReset offsetReset,
 String name,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 Pattern topicPattern)

Deprecated.

Since 4.0. Use addSource(org.apache.kafka.streams.AutoOffsetReset, String, TimestampExtractor, Deserializer, Deserializer, Pattern) instead.

addSource

public <K,
V> Topology addSource(AutoOffsetReset offsetReset,
 String name,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 Pattern topicPattern)

See addSource(String, Pattern).

addSink

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

Add a sink that sends records from upstream processors or sources to the named Kafka topic. The specified topic should be created before the KafkaStreams instance is started.

The sink will use the default values from StreamsConfig for

• key serializer
• value serializer

Furthermore, the producer's configured partitioner is used to write into the topic. If you want to specify a sink specific key or value Serializer, or use a different partitioner, use the corresponding overloaded addSink(...) method.

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 processors or sources, whose output records this sink should consume and write to the specified output topic

Returns:

itself

Throws:

TopologyException - if the provided sink name is not unique, or if a parent processor/source name is unknown or specifies a sink

NullPointerException - if name, topic, or parentNames is null, or parentNames contains a null parent name

See Also:

• addSink(String, TopicNameExtractor, String...)

addSink

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

See addSink(String, String, String...).

addSink

public <K,
V> Topology addSink(String name,
 String topic,
 Serializer<K> keySerializer,
 Serializer<V> valueSerializer,
 String... parentNames)

See addSink(String, String, String...).

addSink

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

See addSink(String, String, String...).

addSink

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

See addSink(String, String, String...).

Takes a TopicNameExtractor (cannot be null) that computes topic names to send records into, instead of a single topic name. The topic name extractor is called for every result record and may compute a different topic name each time. All topics, that the topic name extractor may compute, should be created before the KafkaStreams instance is started. Returning null as topic name is invalid and will result in a runtime exception.

addSink

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

See addSink(String, String, String...).

addSink

public <K,
V> Topology addSink(String name,
 TopicNameExtractor<? super K,? super V> topicExtractor,
 Serializer<K> keySerializer,
 Serializer<V> valueSerializer,
 String... parentNames)

See addSink(String, String, String...).

addSink

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

See addSink(String, String, String...).

addProcessor

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

Add a processor that receives and processed records from one or more parent processors or sources. The Processor can emit any number of result records via ProcessorContext.forward(Record). Any record output by this processor will be forwarded to its child processors and sinks.

By default, the processor is stateless. There is two different state stores, which can be added to the Topology and directly connected to a processor, making the processor stateful:

• state stores for processing (i.e., read/write access)
• read-only state stores

It a (read-only) state store is not directly added to a processing, it can also be connected later. If the supplier provides state stores via ConnectedStoreProvider.stores(), the corresponding StoreBuilders will be added to the topology and connected to this processor automatically. Additionally, even if a processor is stateless, it can still access all global state stores (read-only). There is no need to connect global stores to processors.

All state stores which are connected to a processor and all global stores, can be accessed via context.getStateStore(String) using the context provided via Processor#init():

 public class MyProcessor implements Processor<String, Integer, String, Integer> {
     private ProcessorContext<String, Integer> context;
     private KeyValueStore<String, String> store;

     @Override
     void init(final ProcessorContext<String, Integer> context) {
         this.context = context;
         this.store = context.getStateStore("myStore");
     }

     @Override
     void process(final Record<String, Integer> record) {
         // can access this.context and this.store
     }
 }
 

Furthermore, the provided ProcessorContext gives access to topology, runtime, and record metadata, and allows to schedule punctuations and to request offset commits.

Parameters:

name - the unique name of the processor used to reference this node when adding other processor or sink children

processorSupplier - the supplier used to obtain Processor instances

parentNames - the name of one or more processors or sources, whose output records this processor should receive and process

Returns:

itself

Throws:

TopologyException - if the provided processor name is not unique, or if a parent processor/source name is unknown or specifies a sink

NullPointerException - if name, processorSupplier, or parentNames is null, or parentNames contains a null parent name

See Also:

• ContextualProcessor

addStateStore

public <S extends StateStore> Topology addStateStore(StoreBuilder<S> storeBuilder,
 String... processorNames)

Add a state store to the topology, and optionally connect it to one or more processors. State stores are sharded and the number of shards is determined at runtime by the number of input topic partitions and the structure of the topology. Each connected Processor instance in the topology has access to a single shard of the state store. Additionally, the state store can be accessed from "outside" using the Interactive Queries (IQ) API (cf. KafkaStreams.store(StoreQueryParameters) and KafkaStreams.query(StateQueryRequest)). If you need access to all data in a state store inside a Processor, you can use a (read-only) global state store.

If no processorNames is specified, the state store can be connected to one or more processors later.

Note, if a state store is never connected to any processor, the state store is "dangling" and would not be added to the created ProcessorTopology, when KafkaStreams is started. For this case, the state store is not available for Interactive Queries. If you want to add a state store only for Interactive Queries, you can use a read-only state store.

For failure and recovery, a state store may be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-<storename>-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "storeName" is provided by the store builder, and "-changelog" is a fixed suffix.

You can verify the created ProcessorTopology and added state stores, and retrieve all generated internal topic names, via describe().

Parameters:

storeBuilder - the StoreBuilder used to obtain state store instances (one per shard)

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

Returns:

itself

Throws:

TopologyException - if the state store was already added, or if a processor name is unknown or specifies a source or sink

NullPointerException - if storeBuilder or parentNames is null, or parentNames contains a null parent name

addReadOnlyStateStore

public <K,
V,
S extends StateStore> Topology addReadOnlyStateStore(StoreBuilder<S> storeBuilder,
 String sourceName,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String topic,
 String processorName,
 ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)

Adds a read-only state store to the topology. The state store will be populated with data from the named source topic. State stores are sharded and the number of shards is determined at runtime by the number of input topic partitions for the source topic and the connected processors (if any). Read-only state stores can be accessed from "outside" using the Interactive Queries (IQ) API (cf. KafkaStreams.store(StoreQueryParameters) and KafkaStreams.query(StateQueryRequest)).

The auto.offset.reset property will be set to "earliest" for the source topic. If you want to specify a source specific TimestampExtractor you can use addReadOnlyStateStore(StoreBuilder, String, TimestampExtractor, Deserializer, Deserializer, String, String, ProcessorSupplier).

Connecting a read-only state store to processors is optional. If not connected to any processor, the state store will still be created and can be queried via KafkaStreams.store(StoreQueryParameters) or KafkaStreams.query(StateQueryRequest). If the state store is connected to another processor, each corresponding Processor instance in the topology has read-only access to a single shard of the state store. If you need write access to a state store, you can use a "regular" state store instead. If you need access to all data in a state store inside a Processor, you can use a (read-only) global state store.

The provided ProcessorSupplier will be used to create Processor instances which will be used to process the records from the source topic. These processors are the only ones with write access to the state store, and should contain logic to keep the StateStore up-to-date.

Read-only state stores are always enabled for fault-tolerance and recovery. In contrast to "regular" state stores no dedicated changelog topic will be created in Kafka though, but the source topic is used for recovery. Thus, the source topic should be configured with log compaction.

Parameters:

storeBuilder - the StoreBuilder used to obtain state store instances (one per shard)

sourceName - the unique name of the internally added source

keyDeserializer - the Deserializer for record keys (can be null to use the default key deserializer from StreamsConfig)

valueDeserializer - the Deserializer for record values (can be null to use the default value deserializer from StreamsConfig)

topic - the source topic to read the data from

processorName - the unique name of the internally added processor which maintains the state store

stateUpdateSupplier - the supplier used to obtain Processor instances, which maintain the state store

Returns:

itself

Throws:

TopologyException - if the state store was already added, or if the source or processor names are not unique, or if the source topic has already been registered by another source, read-only state store, or global state store

NullPointerException - if storeBuilder, sourceName, topic, processorName, or stateUpdateSupplier is null

addReadOnlyStateStore

public <K,
V,
S extends StateStore> Topology addReadOnlyStateStore(StoreBuilder<S> storeBuilder,
 String sourceName,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String topic,
 String processorName,
 ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)

See addReadOnlyStateStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier).

addGlobalStore

public <K,
V,
S extends StateStore> Topology addGlobalStore(StoreBuilder<S> storeBuilder,
 String sourceName,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String topic,
 String processorName,
 ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)

Adds a global state store to the topology. The state store will be populated with data from the named source topic. Global state stores are read-only, and contain data from all partitions of the specified source topic. Thus, each KafkaStreams instance has a full copy to the data; the source topic records are effectively broadcast to all instances. In contrast to read-only state stores global state stores are "bootstrapped" on startup, and are maintained by a separate thread. Thus, updates to a global store are not "stream-time synchronized" what may lead to non-deterministic results. Like all other stores, global state stores can be accessed from "outside" using the Interactive Queries (IQ) API) (cf. KafkaStreams.store(StoreQueryParameters) and KafkaStreams.query(StateQueryRequest)).

The auto.offset.reset property will be set to "earliest" for the source topic. If you want to specify a source specific TimestampExtractor you can use addGlobalStore(StoreBuilder, String, TimestampExtractor, Deserializer, Deserializer, String, String, ProcessorSupplier).

All processors of the topology automatically have read-only access to the global store; it is not necessary to connect them. If you need write access to a state store, you can use a "regular" state store instead.

The provided ProcessorSupplier will be used to create Processor instances which will be used to process the records from the source topic. These processors are the only ones with write access to the state store, and should contain logic to keep the StateStore up-to-date.

Global state stores are always enabled for fault-tolerance and recovery. In contrast to "regular" state stores no dedicated changelog topic will be created in Kafka though, but the source topic is used for recovery. Thus, the source topic should be configured with log compaction.

Parameters:

storeBuilder - the StoreBuilder used to obtain the state store (one per KafkaStreams instance)

sourceName - the unique name of the internally added source

keyDeserializer - the Deserializer for record keys (can be null to use the default key deserializer from StreamsConfig)

valueDeserializer - the Deserializer for record values (can be null to use the default value deserializer from StreamsConfig)

topic - the source topic to read the data from

processorName - the unique name of the internally added processor which maintains the state store

stateUpdateSupplier - the supplier used to obtain Processor instances, which maintain the state store

Returns:

itself

Throws:

TopologyException - if the state store was already added, or if the source or processor names are not unique, or if the source topic has already been registered by another source, read-only state store, or global state store

NullPointerException - if storeBuilder, sourceName, topic, processorName, or stateUpdateSupplier is null

addGlobalStore

public <K,
V,
S extends StateStore> Topology addGlobalStore(StoreBuilder<S> storeBuilder,
 String sourceName,
 TimestampExtractor timestampExtractor,
 Deserializer<K> keyDeserializer,
 Deserializer<V> valueDeserializer,
 String topic,
 String processorName,
 ProcessorSupplier<K,V,Void,Void> stateUpdateSupplier)

See addGlobalStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier).

connectProcessorAndStateStores

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

Connect a processor to one or more state stores. The state stores must have been previously added to the topology via addStateStore(StoreBuilder, String...), or addReadOnlyStateStore(StoreBuilder, String, Deserializer, Deserializer, String, String, ProcessorSupplier).

Parameters:

processorName - the name of the processor

stateStoreNames - the names of state stores that the processor should be able to access

Returns:

itself

Throws:

TopologyException - if the processor name or a state store name is unknown, or if the processor name specifies a source or sink

NullPointerException - if processorName or stateStoreNames is null, or if stateStoreNames contains a null state store name

describe

public TopologyDescription describe()

Returns a description of the specified Topology.

Returns:

A description of the topology.