Package org.apache.kafka.streams.kstream

Interface KStream<K,V>

Type Parameters:

K - Type of keys

V - Type of values

@Evolving
public interface KStream<K,V>

KStream is an abstraction of a record stream of KeyValue pairs, i.e., each record is an independent entity/event in the real world. For example a user X might buy two items I1 and I2, and thus there might be two records <K:I1>, <K:I2> in the stream.

A KStream is either defined from one or multiple Kafka topics that are consumed message by message or the result of a KStream transformation. A KTable can also be converted into a KStream.

A KStream can be transformed record by record, joined with another KStream, KTable, GlobalKTable, or can be aggregated into a KTable. Kafka Streams DSL can be mixed-and-matched with Processor API (PAPI) (c.f. Topology) via process(...), transform(...), and transformValues(...).

See Also:

KTable, KGroupedStream, StreamsBuilder.stream(String)

Method Summary

Modifier and Type	Method	Description
KStream<K,V>[]	branch(Predicate<? super K,? super V>... predicates)	Creates an array of KStream from this stream by branching the records in the original stream based on the supplied predicates.
KStream<K,V>	filter(Predicate<? super K,? super V> predicate)	Create a new KStream that consists of all records of this stream which satisfy the given predicate.
KStream<K,V>	filterNot(Predicate<? super K,? super V> predicate)	Create a new KStream that consists all records of this stream which do not satisfy the given predicate.
<KR,VR> KStream<KR,VR>	flatMap(KeyValueMapper<? super K,? super V,? extends java.lang.Iterable<? extends KeyValue<? extends KR,? extends VR>>> mapper)	Transform each record of the input stream into zero or more records in the output stream (both key and value type can be altered arbitrarily).
<VR> KStream<K,VR>	flatMapValues(ValueMapper<? super V,? extends java.lang.Iterable<? extends VR>> mapper)	Create a new KStream by transforming the value of each record in this stream into zero or more values with the same key in the new stream.
<VR> KStream<K,VR>	flatMapValues(ValueMapperWithKey<? super K,? super V,? extends java.lang.Iterable<? extends VR>> mapper)	Create a new KStream by transforming the value of each record in this stream into zero or more values with the same key in the new stream.
void	foreach(ForeachAction<? super K,? super V> action)	Perform an action on each record of KStream.
<KR> KGroupedStream<KR,V>	groupBy(KeyValueMapper<? super K,? super V,KR> selector)	Group the records of this KStream on a new key that is selected using the provided KeyValueMapper and default serializers and deserializers.
<KR> KGroupedStream<KR,V>	groupBy(KeyValueMapper<? super K,? super V,KR> selector, Serde<KR> keySerde, Serde<V> valSerde)	Deprecated. use groupBy(selector, Serialized.with(keySerde, valSerde))
<KR> KGroupedStream<KR,V>	groupBy(KeyValueMapper<? super K,? super V,KR> selector, Serialized<KR,V> serialized)	Group the records of this KStream on a new key that is selected using the provided KeyValueMapper and Serdes as specified by Serialized.
KGroupedStream<K,V>	groupByKey()	Group the records by their current key into a KGroupedStream while preserving the original values and default serializers and deserializers.
KGroupedStream<K,V>	groupByKey(Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use groupByKey(Serialized.with(keySerde, valSerde))
KGroupedStream<K,V>	groupByKey(Serialized<K,V> serialized)	Group the records by their current key into a KGroupedStream while preserving the original values and using the serializers as defined by Serialized.
<GK,GV,RV> KStream<K,RV>	join(GlobalKTable<GK,GV> globalKTable, KeyValueMapper<? super K,? super V,? extends GK> keyValueMapper, ValueJoiner<? super V,? super GV,? extends RV> joiner)	Join records of this stream with GlobalKTable's records using non-windowed inner equi join.
<VO,VR> KStream<K,VR>	join(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)	Join records of this stream with another KStream's records using windowed inner equi join with default serializers and deserializers.
<VO,VR> KStream<K,VR>	join(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V> thisValueSerde, Serde<VO> otherValueSerde)	Deprecated. use join(otherStream, joiner, windows, Joined.with(keySerde, thisValueSerde, otherValueSerde))
<VO,VR> KStream<K,VR>	join(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)	Join records of this stream with another KStream's records using windowed inner equi join with default serializers and deserializers.
<VT,VR> KStream<K,VR>	join(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner)	Join records of this stream with KTable's records using non-windowed inner equi join with default serializers and deserializers.
<VT,VR> KStream<K,VR>	join(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner, Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use join(table, joiner, Joined.with(keySerde, valSerde, null))
<VT,VR> KStream<K,VR>	join(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner, Joined<K,V,VT> joined)	Join records of this stream with KTable's records using non-windowed inner equi join with default serializers and deserializers.
<GK,GV,RV> KStream<K,RV>	leftJoin(GlobalKTable<GK,GV> globalKTable, KeyValueMapper<? super K,? super V,? extends GK> keyValueMapper, ValueJoiner<? super V,? super GV,? extends RV> valueJoiner)	Join records of this stream with GlobalKTable's records using non-windowed left equi join.
<VO,VR> KStream<K,VR>	leftJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)	Join records of this stream with another KStream's records using windowed left equi join with default serializers and deserializers.
<VO,VR> KStream<K,VR>	leftJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V> thisValSerde, Serde<VO> otherValueSerde)	Deprecated. use leftJoin(otherStream, joiner, windows, Joined.with(keySerde, thisValSerde, otherValueSerde))
<VO,VR> KStream<K,VR>	leftJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)	Join records of this stream with another KStream's records using windowed left equi join with default serializers and deserializers.
<VT,VR> KStream<K,VR>	leftJoin(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner)	Join records of this stream with KTable's records using non-windowed left equi join with default serializers and deserializers.
<VT,VR> KStream<K,VR>	leftJoin(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner, Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use leftJoin(table, joiner, Joined.with(keySerde, valSerde, null))
<VT,VR> KStream<K,VR>	leftJoin(KTable<K,VT> table, ValueJoiner<? super V,? super VT,? extends VR> joiner, Joined<K,V,VT> joined)	Join records of this stream with KTable's records using non-windowed left equi join with default serializers and deserializers.
<KR,VR> KStream<KR,VR>	map(KeyValueMapper<? super K,? super V,? extends KeyValue<? extends KR,? extends VR>> mapper)	Transform each record of the input stream into a new record in the output stream (both key and value type can be altered arbitrarily).
<VR> KStream<K,VR>	mapValues(ValueMapper<? super V,? extends VR> mapper)	Transform the value of each input record into a new value (with possible new type) of the output record.
<VR> KStream<K,VR>	mapValues(ValueMapperWithKey<? super K,? super V,? extends VR> mapper)	Transform the value of each input record into a new value (with possible new type) of the output record.
KStream<K,V>	merge(KStream<K,V> stream)	Merge this stream and the given stream into one larger stream.
<VO,VR> KStream<K,VR>	outerJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)	Join records of this stream with another KStream's records using windowed outer equi join with default serializers and deserializers.
<VO,VR> KStream<K,VR>	outerJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V> thisValueSerde, Serde<VO> otherValueSerde)	Deprecated. use outerJoin(otherStream, joiner, windows, Joined.with(keySerde, thisValueSerde, otherValueSerde))
<VO,VR> KStream<K,VR>	outerJoin(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)	Join records of this stream with another KStream's records using windowed outer equi join with default serializers and deserializers.
KStream<K,V>	peek(ForeachAction<? super K,? super V> action)	Perform an action on each record of KStream.
void	print()	Deprecated. use print(Printed)
void	print(java.lang.String label)	Deprecated. use print(Printed.toSysOut()
void	print(Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use print(Printed.toSysOut().withKeyValueMapper(...)
void	print(Serde<K> keySerde, Serde<V> valSerde, java.lang.String label)	Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(...)
void	print(KeyValueMapper<? super K,? super V,java.lang.String> mapper)	Deprecated. use print(Printed.toSysOut().withKeyValueMapper(mapper)
void	print(KeyValueMapper<? super K,? super V,java.lang.String> mapper, java.lang.String label)	Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)
void	print(KeyValueMapper<? super K,? super V,java.lang.String> mapper, Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use print(Printed.toSysOut().withKeyValueMapper(mapper)
void	print(KeyValueMapper<? super K,? super V,java.lang.String> mapper, Serde<K> keySerde, Serde<V> valSerde, java.lang.String label)	Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)
void	print(Printed<K,V> printed)	Print the records of this KStream using the options provided by Printed
void	process(ProcessorSupplier<? super K,? super V> processorSupplier, java.lang.String... stateStoreNames)	Process all records in this stream, one record at a time, by applying a Processor (provided by the given ProcessorSupplier).
<KR> KStream<KR,V>	selectKey(KeyValueMapper<? super K,? super V,? extends KR> mapper)	Set a new key (with possibly new type) for each input record.
KStream<K,V>	through(java.lang.String topic)	Materialize this stream to a topic and creates a new KStream from the topic using default serializers and deserializers and producer's DefaultPartitioner.
KStream<K,V>	through(java.lang.String topic, Produced<K,V> produced)	Materialize this stream to a topic and creates a new KStream from the topic using the Produced instance for configuration of the key serde, value serde, and StreamPartitioner.
KStream<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, java.lang.String topic)	Deprecated. use through(topic, Produced.with(keySerde, valSerde))
KStream<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<? super K,? super V> partitioner, java.lang.String topic)	Deprecated. use through(topic, Produced.with(keySerde, valSerde, partitioner))
KStream<K,V>	through(StreamPartitioner<? super K,? super V> partitioner, java.lang.String topic)	Deprecated. use through(topic, Produced.withStreamPartitioner(partitioner))
void	to(java.lang.String topic)	Materialize this stream to a topic using default serializers specified in the config and producer's DefaultPartitioner.
void	to(java.lang.String topic, Produced<K,V> produced)	Materialize this stream to a topic using the provided Produced instance.
void	to(Serde<K> keySerde, Serde<V> valSerde, java.lang.String topic)	Deprecated. use to(topic, Produced.with(keySerde, valSerde))
void	to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<? super K,? super V> partitioner, java.lang.String topic)	Deprecated. use to(topic, Produced.with(keySerde, valSerde, partitioner)
void	to(StreamPartitioner<? super K,? super V> partitioner, java.lang.String topic)	Deprecated. use to(topic, Produced.withStreamPartitioner(partitioner))
<K1,V1> KStream<K1,V1>	transform(TransformerSupplier<? super K,? super V,KeyValue<K1,V1>> transformerSupplier, java.lang.String... stateStoreNames)	Transform each record of the input stream into zero or more records in the output stream (both key and value type can be altered arbitrarily).
<VR> KStream<K,VR>	transformValues(ValueTransformerSupplier<? super V,? extends VR> valueTransformerSupplier, java.lang.String... stateStoreNames)	Transform the value of each input record into a new value (with possible new type) of the output record.
<VR> KStream<K,VR>	transformValues(ValueTransformerWithKeySupplier<? super K,? super V,? extends VR> valueTransformerSupplier, java.lang.String... stateStoreNames)	Transform the value of each input record into a new value (with possible new type) of the output record.
void	writeAsText(java.lang.String filePath)	Deprecated. use print(Printed.toFile(filePath)
void	writeAsText(java.lang.String filePath, java.lang.String label)	Deprecated. use print(Printed.toFile(filePath).withLabel(label)
void	writeAsText(java.lang.String filePath, java.lang.String label, Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(...)
void	writeAsText(java.lang.String filePath, java.lang.String label, Serde<K> keySerde, Serde<V> valSerde, KeyValueMapper<? super K,? super V,java.lang.String> mapper)	Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)
void	writeAsText(java.lang.String filePath, java.lang.String label, KeyValueMapper<? super K,? super V,java.lang.String> mapper)	Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)
void	writeAsText(java.lang.String filePath, Serde<K> keySerde, Serde<V> valSerde)	Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(...)
void	writeAsText(java.lang.String filePath, Serde<K> keySerde, Serde<V> valSerde, KeyValueMapper<? super K,? super V,java.lang.String> mapper)	Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(mapper)
void	writeAsText(java.lang.String filePath, KeyValueMapper<? super K,? super V,java.lang.String> mapper)	Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(mapper)

Method Detail

filter

KStream<K,V> filter(Predicate<? super K,? super V> predicate)

Create a new KStream that consists of all records of this stream which satisfy the given predicate. All records that do not satisfy the predicate are dropped. This is a stateless record-by-record operation.

Parameters:

predicate - a filter Predicate that is applied to each record

Returns:

a KStream that contains only those records that satisfy the given predicate

See Also:

filterNot(Predicate)

filterNot

KStream<K,V> filterNot(Predicate<? super K,? super V> predicate)

Create a new KStream that consists all records of this stream which do not satisfy the given predicate. All records that do satisfy the predicate are dropped. This is a stateless record-by-record operation.

Parameters:

predicate - a filter Predicate that is applied to each record

Returns:

a KStream that contains only those records that do not satisfy the given predicate

See Also:

filter(Predicate)

selectKey

<KR> KStream<KR,V> selectKey(KeyValueMapper<? super K,? super V,? extends KR> mapper)

Set a new key (with possibly new type) for each input record. The provided KeyValueMapper is applied to each input record and computes a new key for it. Thus, an input record <K,V> can be transformed into an output record <K':V>. This is a stateless record-by-record operation.

For example, you can use this transformation to set a key for a key-less input record <null,V> by extracting a key from the value within your KeyValueMapper. The example below computes the new key as the length of the value string.

 KStream<Byte[], String> keyLessStream = builder.stream("key-less-topic");
 KStream<Integer, String> keyedStream = keyLessStream.selectKey(new KeyValueMapper<Byte[], String, Integer> {
     Integer apply(Byte[] key, String value) {
         return value.length();
     }
 });
 

Setting a new key might result in an internal data redistribution if a key based operator (like an aggregation or join) is applied to the result KStream.

Type Parameters:

KR - the new key type of the result stream

Parameters:

mapper - a KeyValueMapper that computes a new key for each record

Returns:

a KStream that contains records with new key (possibly of different type) and unmodified value

See Also:

map(KeyValueMapper), flatMap(KeyValueMapper), mapValues(ValueMapper), mapValues(ValueMapperWithKey), flatMapValues(ValueMapper), flatMapValues(ValueMapperWithKey)

map

<KR,VR> KStream<KR,VR> map(KeyValueMapper<? super K,? super V,? extends KeyValue<? extends KR,? extends VR>> mapper)

Transform each record of the input stream into a new record in the output stream (both key and value type can be altered arbitrarily). The provided KeyValueMapper is applied to each input record and computes a new output record. Thus, an input record <K,V> can be transformed into an output record <K':V'>. This is a stateless record-by-record operation (cf. transform(TransformerSupplier, String...) for stateful record transformation).

The example below normalizes the String key to upper-case letters and counts the number of token of the value string.

 KStream<String, String> inputStream = builder.stream("topic");
 KStream<String, Integer> outputStream = inputStream.map(new KeyValueMapper<String, String, KeyValue<String, Integer>> {
     KeyValue<String, Integer> apply(String key, String value) {
         return new KeyValue<>(key.toUpperCase(), value.split(" ").length);
     }
 });
 

The provided KeyValueMapper must return a KeyValue type and must not return null.

Mapping records might result in an internal data redistribution if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. mapValues(ValueMapper))

Type Parameters:

KR - the key type of the result stream

VR - the value type of the result stream

Parameters:

mapper - a KeyValueMapper that computes a new output record

Returns:

a KStream that contains records with new key and value (possibly both of different type)

See Also:

selectKey(KeyValueMapper), flatMap(KeyValueMapper), mapValues(ValueMapper), mapValues(ValueMapperWithKey), flatMapValues(ValueMapper), flatMapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

mapValues

<VR> KStream<K,VR> mapValues(ValueMapper<? super V,? extends VR> mapper)

Transform the value of each input record into a new value (with possible new type) of the output record. The provided ValueMapper is applied to each input record value and computes a new value for it. Thus, an input record <K,V> can be transformed into an output record <K:V'>. This is a stateless record-by-record operation (cf. transformValues(ValueTransformerSupplier, String...) for stateful value transformation).

The example below counts the number of token of the value string.

 KStream<String, String> inputStream = builder.stream("topic");
 KStream<String, Integer> outputStream = inputStream.mapValues(new ValueMapper<String, Integer> {
     Integer apply(String value) {
         return value.split(" ").length;
     }
 });
 

Setting a new value preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. map(KeyValueMapper))

Type Parameters:

VR - the value type of the result stream

Parameters:

mapper - a ValueMapper that computes a new output value

Returns:

a KStream that contains records with unmodified key and new values (possibly of different type)

See Also:

selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), flatMapValues(ValueMapper), flatMapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

mapValues

<VR> KStream<K,VR> mapValues(ValueMapperWithKey<? super K,? super V,? extends VR> mapper)

Transform the value of each input record into a new value (with possible new type) of the output record. The provided ValueMapperWithKey is applied to each input record value and computes a new value for it. Thus, an input record <K,V> can be transformed into an output record <K:V'>. This is a stateless record-by-record operation (cf. transformValues(ValueTransformerSupplier, String...) for stateful value transformation).

The example below counts the number of tokens of key and value strings.

 KStream<String, String> inputStream = builder.stream("topic");
 KStream<String, Integer> outputStream = inputStream.mapValues(new ValueMapperWithKey<String, String, Integer> {
     Integer apply(String readOnlyKey, String value) {
         return readOnlyKey.split(" ").length + value.split(" ").length;
     }
 });
 

Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning. So, setting a new value preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. map(KeyValueMapper))

Type Parameters:

VR - the value type of the result stream

Parameters:

mapper - a ValueMapperWithKey that computes a new output value

Returns:

a KStream that contains records with unmodified key and new values (possibly of different type)

See Also:

selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), flatMapValues(ValueMapper), flatMapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

flatMap

<KR,VR> KStream<KR,VR> flatMap(KeyValueMapper<? super K,? super V,? extends java.lang.Iterable<? extends KeyValue<? extends KR,? extends VR>>> mapper)

Transform each record of the input stream into zero or more records in the output stream (both key and value type can be altered arbitrarily). The provided KeyValueMapper is applied to each input record and computes zero or more output records. Thus, an input record <K,V> can be transformed into output records <K':V'>, <K'':V''>, .... This is a stateless record-by-record operation (cf. transform(TransformerSupplier, String...) for stateful record transformation).

The example below splits input records <null:String> containing sentences as values into their words and emit a record <word:1> for each word.

 KStream<byte[], String> inputStream = builder.stream("topic");
 KStream<String, Integer> outputStream = inputStream.flatMap(new KeyValueMapper<byte[], String, Iterable<KeyValue<String, Integer>>> {
     Iterable<KeyValue<String, Integer>> apply(byte[] key, String value) {
         String[] tokens = value.split(" ");
         List<KeyValue<String, Integer>> result = new ArrayList<>(tokens.length);

         for(String token : tokens) {
             result.add(new KeyValue<>(token, 1));
         }

         return result;
     }
 });
 

The provided KeyValueMapper must return an Iterable (e.g., any Collection type) and the return value must not be null.

Flat-mapping records might result in an internal data redistribution if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. flatMapValues(ValueMapper))

Type Parameters:

KR - the key type of the result stream

VR - the value type of the result stream

Parameters:

mapper - a KeyValueMapper that computes the new output records

Returns:

a KStream that contains more or less records with new key and value (possibly of different type)

See Also:

selectKey(KeyValueMapper), map(KeyValueMapper), mapValues(ValueMapper), mapValues(ValueMapperWithKey), flatMapValues(ValueMapper), flatMapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

flatMapValues

<VR> KStream<K,VR> flatMapValues(ValueMapper<? super V,? extends java.lang.Iterable<? extends VR>> mapper)

Create a new KStream by transforming the value of each record in this stream into zero or more values with the same key in the new stream. Transform the value of each input record into zero or more records with the same (unmodified) key in the output stream (value type can be altered arbitrarily). The provided ValueMapper is applied to each input record and computes zero or more output values. Thus, an input record <K,V> can be transformed into output records <K:V'>, <K:V''>, .... This is a stateless record-by-record operation (cf. transformValues(ValueTransformerSupplier, String...) for stateful value transformation).

The example below splits input records <null:String> containing sentences as values into their words.

 KStream<byte[], String> inputStream = builder.stream("topic");
 KStream<byte[], String> outputStream = inputStream.flatMapValues(new ValueMapper<String, Iterable<String>> {
     Iterable<String> apply(String value) {
         return Arrays.asList(value.split(" "));
     }
 });
 

The provided ValueMapper must return an Iterable (e.g., any Collection type) and the return value must not be null.

Splitting a record into multiple records with the same key preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. flatMap(KeyValueMapper))

Type Parameters:

VR - the value type of the result stream

Parameters:

mapper - a ValueMapper the computes the new output values

Returns:

a KStream that contains more or less records with unmodified keys and new values of different type

See Also:

selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), mapValues(ValueMapper), mapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

flatMapValues

<VR> KStream<K,VR> flatMapValues(ValueMapperWithKey<? super K,? super V,? extends java.lang.Iterable<? extends VR>> mapper)

Create a new KStream by transforming the value of each record in this stream into zero or more values with the same key in the new stream. Transform the value of each input record into zero or more records with the same (unmodified) key in the output stream (value type can be altered arbitrarily). The provided ValueMapperWithKey is applied to each input record and computes zero or more output values. Thus, an input record <K,V> can be transformed into output records <K:V'>, <K:V''>, .... This is a stateless record-by-record operation (cf. transformValues(ValueTransformerSupplier, String...) for stateful value transformation).

The example below splits input records <Integer:String>, with key=1, containing sentences as values into their words.

 KStream<Integer, String> inputStream = builder.stream("topic");
 KStream<Integer, String> outputStream = inputStream.flatMapValues(new ValueMapper<Integer, String, Iterable<String>> {
     Iterable<Integer, String> apply(Integer readOnlyKey, String value) {
         if(readOnlyKey == 1) {
             return Arrays.asList(value.split(" "));
         } else {
             return Arrays.asList(value);
         }
     }
 });
 

The provided ValueMapperWithKey must return an Iterable (e.g., any Collection type) and the return value must not be null.

Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning. So, splitting a record into multiple records with the same key preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. flatMap(KeyValueMapper))

Type Parameters:

VR - the value type of the result stream

Parameters:

mapper - a ValueMapperWithKey the computes the new output values

Returns:

a KStream that contains more or less records with unmodified keys and new values of different type

See Also:

selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), mapValues(ValueMapper), mapValues(ValueMapperWithKey), transform(TransformerSupplier, String...), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...)

print

@Deprecated
void print()

Deprecated. use print(Printed)

Print the records of this stream to System.out. This function will use the generated name of the parent processor node to label the key/value pairs printed to the console.

The default serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

print

@Deprecated
void print(java.lang.String label)

Deprecated. use print(Printed.toSysOut()

Print the records of this stream to System.out. This function will use the given name to label the key/value pairs printed to the console.

The default serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

label - the name used to label the key/value pairs printed to the console

print

@Deprecated
void print(Serde<K> keySerde,
           Serde<V> valSerde)

Deprecated. use print(Printed.toSysOut().withKeyValueMapper(...)

Print the records of this stream to System.out. This function will use the generated name of the parent processor node to label the key/value pairs printed to the console.

The provided serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

keySerde - key serde used to deserialize key if type is byte[],

valSerde - value serde used to deserialize value if type is byte[],

print

@Deprecated
void print(Serde<K> keySerde,
           Serde<V> valSerde,
           java.lang.String label)

Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(...)

Print the records of this stream to System.out.

The provided serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

keySerde - key serde used to deserialize key if type is byte[],

valSerde - value serde used to deserialize value if type is byte[],

label - the name used to label the key/value pairs printed to the console

print

@Deprecated
void print(KeyValueMapper<? super K,? super V,java.lang.String> mapper)

Deprecated. use print(Printed.toSysOut().withKeyValueMapper(mapper)

Print the customized output with System.out.

The default serde will be use to deserialize key or value if type is byte[]. The user provided KeyValueMapper which customizes output is used to print with System.out

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Parameters:

mapper - a KeyValueMapper that computes output type String.

print

@Deprecated
void print(KeyValueMapper<? super K,? super V,java.lang.String> mapper,
           java.lang.String label)

Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)

Print the customized output with System.out.

The default serde will be used to deserialize key or value if type is byte[]. The user provided KeyValueMapper which customizes output is used to print with System.out

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Parameters:

mapper - a KeyValueMapper that computes output type String.

label - The given name which labels output will be printed.

print

@Deprecated
void print(KeyValueMapper<? super K,? super V,java.lang.String> mapper,
           Serde<K> keySerde,
           Serde<V> valSerde)

Deprecated. use print(Printed.toSysOut().withKeyValueMapper(mapper)

Print the customized output with System.out.

The user provided KeyValueMapper which customizes output is used to print with System.out The provided serde will be use to deserialize key or value if type is byte[].

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The provided KeyValueMapper's mapped value type must be String.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

mapper - a KeyValueMapper that computes output type String.

keySerde - a Serde used to deserialize key if type is byte[].

valSerde - a Serde used to deserialize value if type is byte[].

print

@Deprecated
void print(KeyValueMapper<? super K,? super V,java.lang.String> mapper,
           Serde<K> keySerde,
           Serde<V> valSerde,
           java.lang.String label)

Deprecated. use print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)

Print the customized output with System.out.

The user provided KeyValueMapper which customizes output is used to print with System.out. The provided serde will be use to deserialize key or value if type is byte[].

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The provided KeyValueMapper's mapped value type must be String.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

mapper - a KeyValueMapper that computes output type String.

keySerde - a Serde used to deserialize key if type is byte[].

valSerde - a Serde used to deserialize value if type is byte[].

label - The given name which labels output will be printed.

print

void print(Printed<K,V> printed)

Print the records of this KStream using the options provided by Printed

Parameters:

printed - options for printing

merge

KStream<K,V> merge(KStream<K,V> stream)

Merge this stream and the given stream into one larger stream.

There is no ordering guarantee between records from this KStream and records from the provided KStream in the merged stream. Relative order is preserved within each input stream though (ie, records within one input stream are processed in order).

Parameters:

stream - a stream which is to be merged into this stream

Returns:

a merged stream containing all records from this and the provided KStream

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath)

Deprecated. use print(Printed.toFile(filePath)

Write the records of this stream to a file at the given path. This function will use the generated name of the parent processor node to label the key/value pairs printed to the file.

The default serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - name of the file to write to

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 java.lang.String label)

Deprecated. use print(Printed.toFile(filePath).withLabel(label)

Write the records of this stream to a file at the given path. This function will use the given name to label the key/value printed to the file.

The default serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - name of the file to write to

label - the name used to label the key/value pairs written to the file

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 Serde<K> keySerde,
                 Serde<V> valSerde)

Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(...)

Write the records of this stream to a file at the given path. This function will use the generated name of the parent processor node to label the key/value pairs printed to the file.

The provided serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - name of the file to write to

keySerde - key serde used to deserialize key if type is byte[],

valSerde - value serde used to deserialize value if type is byte[],

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 java.lang.String label,
                 Serde<K> keySerde,
                 Serde<V> valSerde)

Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(...)

Write the records of this stream to a file at the given path. This function will use the given name to label the key/value printed to the file.

The provided serde will be used to deserialize the key or value in case the type is byte[] before calling toString() on the deserialized object.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - name of the file to write to

label - the name used to label the key/value pairs written to the file

keySerde - key serde used to deserialize key if type is byte[],

valSerde - value serde used deserialize value if type is byte[],

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 KeyValueMapper<? super K,? super V,java.lang.String> mapper)

Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(mapper)

Write the customised output to a given file path.

The user provided KeyValueMapper which customizes output is used to write to file. This function will use default name of stream to label records.

The default key and value serde will used to deserialize byte[] records before calling toString().

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Parameters:

filePath - path of the file to write to.

mapper - a KeyValueMapper that computes output type String.

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 java.lang.String label,
                 KeyValueMapper<? super K,? super V,java.lang.String> mapper)

Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)

Write the customised output to a given file path.

The user provided KeyValueMapper which customizes output is used to write to file. This function will use given name of stream to label records.

The default key and value serde will used to deserialize byte[] records before calling toString().

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Parameters:

filePath - path of the file to write to.

label - the name used to label records written to file.

mapper - a KeyValueMapper that computes output type String.

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 Serde<K> keySerde,
                 Serde<V> valSerde,
                 KeyValueMapper<? super K,? super V,java.lang.String> mapper)

Deprecated. use print(Printed.toFile(filePath).withKeyValueMapper(mapper)

Write the customised output to a given file path.

The user provided KeyValueMapper which customizes output is used to write to file. This function will use default name of stream to label records.

The given key and value serde will be used to deserialize byte[] records before calling toString().

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - path of the file to write to.

keySerde - key serde used to deserialize key if type is byte[].

valSerde - value serde used to deserialize value if type is byte[].

mapper - a KeyValueMapper that computes output type String.

writeAsText

@Deprecated
void writeAsText(java.lang.String filePath,
                 java.lang.String label,
                 Serde<K> keySerde,
                 Serde<V> valSerde,
                 KeyValueMapper<? super K,? super V,java.lang.String> mapper)

Deprecated. use print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)

Write the customised output to a given file path.

The user provided KeyValueMapper which customizes output is used to write to file. This function will use given name of stream to label records.

The given key and value serde will be used to deserialize byte[] records before calling toString().

The example below shows the way to customize output data.

 final KeyValueMapper<Integer, String, String> mapper = new KeyValueMapper<Integer, String, String>() {
     public String apply(Integer key, String value) {
         return String.format("(%d, %s)", key, value);
     }
 };
 

The KeyValueMapper's mapped value type must be String.

Implementors will need to override toString() for keys and values that are not of type String, Integer etc. to get meaningful information.

Parameters:

filePath - path of the file to write to.

label - the name used to label records written to file.

keySerde - key serde used to deserialize key if type is byte[].

valSerde - value serde used to deserialize value if type is byte[].

mapper - a KeyValueMapper that computes output type String.

foreach

void foreach(ForeachAction<? super K,? super V> action)

Perform an action on each record of KStream. This is a stateless record-by-record operation (cf. process(ProcessorSupplier, String...)). Note that this is a terminal operation that returns void.

Parameters:

action - an action to perform on each record

See Also:

process(ProcessorSupplier, String...)

peek

KStream<K,V> peek(ForeachAction<? super K,? super V> action)

Perform an action on each record of KStream. This is a stateless record-by-record operation (cf. process(ProcessorSupplier, String...)).

Peek is a non-terminal operation that triggers a side effect (such as logging or statistics collection) and returns an unchanged stream.

Note that since this operation is stateless, it may execute multiple times for a single record in failure cases.

Parameters:

action - an action to perform on each record

See Also:

process(ProcessorSupplier, String...)

branch

KStream<K,V>[] branch(Predicate<? super K,? super V>... predicates)

Creates an array of KStream from this stream by branching the records in the original stream based on the supplied predicates. Each record is evaluated against the supplied predicates, and predicates are evaluated in order. Each stream in the result array corresponds position-wise (index) to the predicate in the supplied predicates. The branching happens on first-match: A record in the original stream is assigned to the corresponding result stream for the first predicate that evaluates to true, and is assigned to this stream only. A record will be dropped if none of the predicates evaluate to true. This is a stateless record-by-record operation.

Parameters:

predicates - the ordered list of Predicate instances

Returns:

multiple distinct substreams of this KStream

through

KStream<K,V> through(java.lang.String topic)

Materialize this stream to a topic and creates a new KStream from the topic using default serializers and deserializers and producer's DefaultPartitioner. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

This is equivalent to calling #to(someTopicName) and StreamsBuilder#stream(someTopicName).

Parameters:

topic - the topic name

Returns:

a KStream that contains the exact same (and potentially repartitioned) records as this KStream

through

@Deprecated
KStream<K,V> through(StreamPartitioner<? super K,? super V> partitioner,
                     java.lang.String topic)

Deprecated. use through(topic, Produced.withStreamPartitioner(partitioner))

Materialize this stream to a topic and creates a new KStream from the topic using default serializers and deserializers and a customizable StreamPartitioner to determine the distribution of records to partitions. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

This is equivalent to calling #to(StreamPartitioner, someTopicName) and StreamsBuilder#stream(someTopicName).

Parameters:

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified producer's DefaultPartitioner will be used

topic - the topic name

Returns:

a KStream that contains the exact same (and potentially repartitioned) records as this KStream

through

@Deprecated
KStream<K,V> through(Serde<K> keySerde,
                     Serde<V> valSerde,
                     java.lang.String topic)

Deprecated. use through(topic, Produced.with(keySerde, valSerde))

Materialize this stream to a topic, and creates a new KStream from the topic. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

If keySerde provides a WindowedSerializer for the key WindowedStreamPartitioner is used—otherwise producer's DefaultPartitioner is used.

This is equivalent to calling #to(keySerde, valSerde, someTopicName) and KStreamBuilder#stream(keySerde, valSerde, someTopicName).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default key serde defined in the configuration will be used

valSerde - value serde used to send key-value pairs, if not specified the default value serde defined in the configuration will be used

topic - the topic name

Returns:

a KStream that contains the exact same (and potentially repartitioned) records as this KStream

through

@Deprecated
KStream<K,V> through(Serde<K> keySerde,
                     Serde<V> valSerde,
                     StreamPartitioner<? super K,? super V> partitioner,
                     java.lang.String topic)

Deprecated. use through(topic, Produced.with(keySerde, valSerde, partitioner))

Materialize this stream to a topic and creates a new KStream from the topic using a customizable StreamPartitioner to determine the distribution of records to partitions. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

This is equivalent to calling #to(keySerde, valSerde, StreamPartitioner, someTopicName) and KStreamBuilder#stream(keySerde, valSerde, someTopicName).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default key serde defined in the configuration will be used

valSerde - value serde used to send key-value pairs, if not specified the default value serde defined in the configuration will be used

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified and keySerde provides a WindowedSerializer for the key WindowedStreamPartitioner will be used—otherwise DefaultPartitioner will be used

topic - the topic name

Returns:

a KStream that contains the exact same (and potentially repartitioned) records as this KStream

through

KStream<K,V> through(java.lang.String topic,
                     Produced<K,V> produced)

Materialize this stream to a topic and creates a new KStream from the topic using the Produced instance for configuration of the key serde, value serde, and StreamPartitioner. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

This is equivalent to calling to(someTopic, Produced.with(keySerde, valueSerde) and StreamsBuilder#stream(someTopicName, Consumed.with(keySerde, valueSerde)).

Parameters:

topic -

produced -

Returns:

a KStream that contains the exact same (and potentially repartitioned) records as this KStream

to

void to(java.lang.String topic)

Materialize this stream to a topic using default serializers specified in the config and producer's DefaultPartitioner. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

Parameters:

topic - the topic name

to

@Deprecated
void to(StreamPartitioner<? super K,? super V> partitioner,
        java.lang.String topic)

Deprecated. use to(topic, Produced.withStreamPartitioner(partitioner))

Materialize this stream to a topic using default serializers specified in the config and a customizable StreamPartitioner to determine the distribution of records to partitions. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

Parameters:

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified producer's DefaultPartitioner will be used

topic - the topic name

to

@Deprecated
void to(Serde<K> keySerde,
        Serde<V> valSerde,
        java.lang.String topic)

Deprecated. use to(topic, Produced.with(keySerde, valSerde))

Materialize this stream to a topic. If keySerde provides a WindowedSerializer for the key WindowedStreamPartitioner is used—otherwise producer's DefaultPartitioner is used. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

topic - the topic name

to

@Deprecated
void to(Serde<K> keySerde,
        Serde<V> valSerde,
        StreamPartitioner<? super K,? super V> partitioner,
        java.lang.String topic)

Deprecated. use to(topic, Produced.with(keySerde, valSerde, partitioner)

Materialize this stream to a topic using a customizable StreamPartitioner to determine the distribution of records to partitions. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified and keySerde provides a WindowedSerializer for the key WindowedStreamPartitioner will be used—otherwise DefaultPartitioner will be used

topic - the topic name

to

void to(java.lang.String topic,
        Produced<K,V> produced)

Materialize this stream to a topic using the provided Produced instance. The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is started).

Parameters:

produced - the options to use when producing to the topic

topic - the topic name

transform

<K1,V1> KStream<K1,V1> transform(TransformerSupplier<? super K,? super V,KeyValue<K1,V1>> transformerSupplier,
                                 java.lang.String... stateStoreNames)

Transform each record of the input stream into zero or more records in the output stream (both key and value type can be altered arbitrarily). A Transformer (provided by the given TransformerSupplier) is applied to each input record and computes zero or more output records. Thus, an input record <K,V> can be transformed into output records <K':V'>, <K'':V''>, .... This is a stateful record-by-record operation (cf. flatMap(KeyValueMapper)). Furthermore, via Punctuator.punctuate(long) the processing progress can be observed and additional periodic actions can be performed.

In order to assign a state, the state must be created and registered beforehand:

 // create store
 StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
         Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myTransformState"),
                 Serdes.String(),
                 Serdes.String());
 // register store
 builder.addStateStore(keyValueStoreBuilder);

 KStream outputStream = inputStream.transform(new TransformerSupplier() { ... }, "myTransformState");
 

Within the Transformer, the state is obtained via the ProcessorContext. To trigger periodic actions via punctuate(), a schedule must be registered. The Transformer must return a KeyValue type in transform() and punctuate().

 new TransformerSupplier() {
     Transformer get() {
         return new Transformer() {
             private ProcessorContext context;
             private StateStore state;

             void init(ProcessorContext context) {
                 this.context = context;
                 this.state = context.getStateStore("myTransformState");
                 // punctuate each 1000ms; can access this.state
                 // can emit as many new KeyValue pairs as required via this.context#forward()
                 context.schedule(1000, PunctuationType.WALL_CLOCK_TIME, new Punctuator(..));
             }

             KeyValue transform(K key, V value) {
                 // can access this.state
                 // can emit as many new KeyValue pairs as required via this.context#forward()
                 return new KeyValue(key, value); // can emit a single value via return -- can also be null
             }

             void close() {
                 // can access this.state
                 // can emit as many new KeyValue pairs as required via this.context#forward()
             }
         }
     }
 }
 

Transforming records might result in an internal data redistribution if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. transformValues(ValueTransformerSupplier, String...))

Type Parameters:

K1 - the key type of the new stream

V1 - the value type of the new stream

Parameters:

transformerSupplier - a instance of TransformerSupplier that generates a Transformer

stateStoreNames - the names of the state stores used by the processor

Returns:

a KStream that contains more or less records with new key and value (possibly of different type)

See Also:

flatMap(KeyValueMapper), transformValues(ValueTransformerSupplier, String...), transformValues(ValueTransformerWithKeySupplier, String...), process(ProcessorSupplier, String...)

transformValues

<VR> KStream<K,VR> transformValues(ValueTransformerSupplier<? super V,? extends VR> valueTransformerSupplier,
                                   java.lang.String... stateStoreNames)

Transform the value of each input record into a new value (with possible new type) of the output record. A ValueTransformer (provided by the given ValueTransformerSupplier) is applies to each input record value and computes a new value for it. Thus, an input record <K,V> can be transformed into an output record <K:V'>. This is a stateful record-by-record operation (cf. mapValues(ValueMapper)). Furthermore, via Punctuator.punctuate(long) the processing progress can be observed and additional periodic actions get be performed.

In order to assign a state, the state must be created and registered beforehand:

 // create store
 StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
         Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
                 Serdes.String(),
                 Serdes.String());
 // register store
 builder.addStateStore(keyValueStoreBuilder);

 KStream outputStream = inputStream.transformValues(new ValueTransformerSupplier() { ... }, "myValueTransformState");
 

Within the ValueTransformer, the state is obtained via the ProcessorContext. To trigger periodic actions via punctuate(), a schedule must be registered. In contrast to transform(), no additional KeyValue pairs should be emitted via ProcessorContext.forward().

 new ValueTransformerSupplier() {
     ValueTransformer get() {
         return new ValueTransformer() {
             private StateStore state;

             void init(ProcessorContext context) {
                 this.state = context.getStateStore("myValueTransformState");
                 context.schedule(1000, PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
             }

             NewValueType transform(V value) {
                 // can access this.state
                 return new NewValueType(); // or null
             }

             void close() {
                 // can access this.state
             }
         }
     }
 }
 

Setting a new value preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. transform(TransformerSupplier, String...))

Type Parameters:

VR - the value type of the result stream

Parameters:

valueTransformerSupplier - a instance of ValueTransformerSupplier that generates a ValueTransformer

stateStoreNames - the names of the state stores used by the processor

Returns:

a KStream that contains records with unmodified key and new values (possibly of different type)

See Also:

mapValues(ValueMapper), mapValues(ValueMapperWithKey), transform(TransformerSupplier, String...)

transformValues

<VR> KStream<K,VR> transformValues(ValueTransformerWithKeySupplier<? super K,? super V,? extends VR> valueTransformerSupplier,
                                   java.lang.String... stateStoreNames)

Transform the value of each input record into a new value (with possible new type) of the output record. A ValueTransformerWithKey (provided by the given ValueTransformerWithKeySupplier) is applies to each input record value and computes a new value for it. Thus, an input record <K,V> can be transformed into an output record <K:V'>. This is a stateful record-by-record operation (cf. mapValues(ValueMapperWithKey)). Furthermore, via Punctuator.punctuate(long) the processing progress can be observed and additional periodic actions get be performed.

In order to assign a state, the state must be created and registered beforehand:

 // create store
 StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
         Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
                 Serdes.String(),
                 Serdes.String());
 // register store
 builder.addStateStore(keyValueStoreBuilder);

 KStream outputStream = inputStream.transformValues(new ValueTransformerWithKeySupplier() { ... }, "myValueTransformState");
 

Within the ValueTransformerWithKey, the state is obtained via the ProcessorContext. To trigger periodic actions via punctuate(), a schedule must be registered. In contrast to transform(), no additional KeyValue pairs should be emitted via ProcessorContext.forward().

 new ValueTransformerWithKeySupplier() {
     ValueTransformerWithKey get() {
         return new ValueTransformerWithKey() {
             private StateStore state;

             void init(ProcessorContext context) {
                 this.state = context.getStateStore("myValueTransformState");
                 context.schedule(1000, PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
             }

             NewValueType transform(K readOnlyKey, V value) {
                 // can access this.state and use read-only key
                 return new NewValueType(readOnlyKey); // or null
             }

             void close() {
                 // can access this.state
             }
         }
     }
 }
 

Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning. So, setting a new value preserves data co-location with respect to the key. Thus, no internal data redistribution is required if a key based operator (like an aggregation or join) is applied to the result KStream. (cf. transform(TransformerSupplier, String...))

Type Parameters:

VR - the value type of the result stream

Parameters:

valueTransformerSupplier - a instance of ValueTransformerWithKeySupplier that generates a ValueTransformerWithKey

stateStoreNames - the names of the state stores used by the processor

Returns:

a KStream that contains records with unmodified key and new values (possibly of different type)

See Also:

mapValues(ValueMapper), mapValues(ValueMapperWithKey), transform(TransformerSupplier, String...)

process

void process(ProcessorSupplier<? super K,? super V> processorSupplier,
             java.lang.String... stateStoreNames)

Process all records in this stream, one record at a time, by applying a Processor (provided by the given ProcessorSupplier). This is a stateful record-by-record operation (cf. foreach(ForeachAction)). Furthermore, via Punctuator.punctuate(long) the processing progress can be observed and additional periodic actions can be performed. Note that this is a terminal operation that returns void.

In order to assign a state, the state must be created and registered beforehand:

 // create store
 StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
         Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myProcessorState"),
                 Serdes.String(),
                 Serdes.String());
 // register store
 builder.addStateStore(keyValueStoreBuilder);

 inputStream.process(new ProcessorSupplier() { ... }, "myProcessorState");
 

Within the Processor, the state is obtained via the ProcessorContext. To trigger periodic actions via punctuate(), a schedule must be registered.

 new ProcessorSupplier() {
     Processor get() {
         return new Processor() {
             private StateStore state;

             void init(ProcessorContext context) {
                 this.state = context.getStateStore("myProcessorState");
                 context.schedule(1000, PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
             }

             void process(K key, V value) {
                 // can access this.state
             }

             void close() {
                 // can access this.state
             }
         }
     }
 }
 

Parameters:

processorSupplier - a instance of ProcessorSupplier that generates a Processor

stateStoreNames - the names of the state store used by the processor

See Also:

foreach(ForeachAction), transform(TransformerSupplier, String...)

groupByKey

KGroupedStream<K,V> groupByKey()

Group the records by their current key into a KGroupedStream while preserving the original values and default serializers and deserializers. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). If a record key is null the record will not be included in the resulting KGroupedStream.

If a key changing operator was used before this operation (e.g., selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), or transform(TransformerSupplier, String...)), and no data redistribution happened afterwards (e.g., via through(String)) an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

For this case, all data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned correctly on its key. If the last key changing operator changed the key type, it is recommended to use groupByKey(Serialized) instead.

Returns:

a KGroupedStream that contains the grouped records of the original KStream

See Also:

groupBy(KeyValueMapper)

groupByKey

KGroupedStream<K,V> groupByKey(Serialized<K,V> serialized)

Group the records by their current key into a KGroupedStream while preserving the original values and using the serializers as defined by Serialized. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). If a record key is null the record will not be included in the resulting KGroupedStream.

If a key changing operator was used before this operation (e.g., selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), or transform(TransformerSupplier, String...)), and no data redistribution happened afterwards (e.g., via through(String)) an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

For this case, all data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned correctly on its key.

Returns:

a KGroupedStream that contains the grouped records of the original KStream

See Also:

groupBy(KeyValueMapper)

groupByKey

@Deprecated
KGroupedStream<K,V> groupByKey(Serde<K> keySerde,
                               Serde<V> valSerde)

Deprecated. use groupByKey(Serialized.with(keySerde, valSerde))

Group the records by their current key into a KGroupedStream while preserving the original values. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). If a record key is null the record will not be included in the resulting KGroupedStream.

If a key changing operator was used before this operation (e.g., selectKey(KeyValueMapper), map(KeyValueMapper), flatMap(KeyValueMapper), or transform(TransformerSupplier, String...)), and no data redistribution happened afterwards (e.g., via through(String)) an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

For this case, all data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned correctly on its key.

Parameters:

keySerde - key serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

valSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

Returns:

a KGroupedStream that contains the grouped records of the original KStream

groupBy

<KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector)

Group the records of this KStream on a new key that is selected using the provided KeyValueMapper and default serializers and deserializers. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). The KeyValueMapper selects a new key (with should be of the same type) while preserving the original values. If the new record key is null the record will not be included in the resulting KGroupedStream

Because a new key is selected, an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

All data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned on the new key.

This operation is equivalent to calling selectKey(KeyValueMapper) followed by groupByKey(). If the key type is changed, it is recommended to use groupBy(KeyValueMapper, Serialized) instead.

Type Parameters:

KR - the key type of the result KGroupedStream

Parameters:

selector - a KeyValueMapper that computes a new key for grouping

Returns:

a KGroupedStream that contains the grouped records of the original KStream

groupBy

<KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector,
                                  Serialized<KR,V> serialized)

Group the records of this KStream on a new key that is selected using the provided KeyValueMapper and Serdes as specified by Serialized. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). The KeyValueMapper selects a new key (with should be of the same type) while preserving the original values. If the new record key is null the record will not be included in the resulting KGroupedStream.

Because a new key is selected, an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

All data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned on the new key.

This operation is equivalent to calling selectKey(KeyValueMapper) followed by groupByKey().

Type Parameters:

KR - the key type of the result KGroupedStream

Parameters:

selector - a KeyValueMapper that computes a new key for grouping

Returns:

a KGroupedStream that contains the grouped records of the original KStream

groupBy

@Deprecated
<KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector,
                                  Serde<KR> keySerde,
                                  Serde<V> valSerde)

Deprecated. use groupBy(selector, Serialized.with(keySerde, valSerde))

Group the records of this KStream on a new key that is selected using the provided KeyValueMapper. Grouping a stream on the record key is required before an aggregation operator can be applied to the data (cf. KGroupedStream). The KeyValueMapper selects a new key (with potentially different type) while preserving the original values. If the new record key is null the record will not be included in the resulting KGroupedStream.

Because a new key is selected, an internal repartitioning topic will be created in Kafka. This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

All data of this stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the resulting KGroupedStream is partitioned on the new key.

This is equivalent to calling selectKey(KeyValueMapper) followed by groupByKey(Serde, Serde).

Type Parameters:

KR - the key type of the result KGroupedStream

Parameters:

selector - a KeyValueMapper that computes a new key for grouping

keySerde - key serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

valSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

Returns:

a KGroupedStream that contains the grouped records of the original KStream

See Also:

groupByKey()

join

<VO,VR> KStream<K,VR> join(KStream<K,VO> otherStream,
                           ValueJoiner<? super V,? super VO,? extends VR> joiner,
                           JoinWindows windows)

Join records of this stream with another KStream's records using windowed inner equi join with default serializers and deserializers. The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key and within the joining window intervals

See Also:

leftJoin(KStream, ValueJoiner, JoinWindows), outerJoin(KStream, ValueJoiner, JoinWindows)

join

<VO,VR> KStream<K,VR> join(KStream<K,VO> otherStream,
                           ValueJoiner<? super V,? super VO,? extends VR> joiner,
                           JoinWindows windows,
                           Joined<K,V,VO> joined)

Join records of this stream with another KStream's records using windowed inner equi join with default serializers and deserializers. The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

joined - a Joined instance that defines the serdes to be used to serialize/deserialize inputs and outputs of the joined streams

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key and within the joining window intervals

See Also:

leftJoin(KStream, ValueJoiner, JoinWindows, Joined), outerJoin(KStream, ValueJoiner, JoinWindows, Joined)

join

@Deprecated
<VO,VR> KStream<K,VR> join(KStream<K,VO> otherStream,
                           ValueJoiner<? super V,? super VO,? extends VR> joiner,
                           JoinWindows windows,
                           Serde<K> keySerde,
                           Serde<V> thisValueSerde,
                           Serde<VO> otherValueSerde)

Deprecated. use join(otherStream, joiner, windows, Joined.with(keySerde, thisValueSerde, otherValueSerde))

Join records of this stream with another KStream's records using windowed inner equi join. The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

keySerde - key serdes for materializing both streams, if not specified the default serdes defined in the configs will be used

thisValueSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key and within the joining window intervals

See Also:

leftJoin(KStream, ValueJoiner, JoinWindows, Joined), outerJoin(KStream, ValueJoiner, JoinWindows, Joined)

leftJoin

<VO,VR> KStream<K,VR> leftJoin(KStream<K,VO> otherStream,
                               ValueJoiner<? super V,? super VO,? extends VR> joiner,
                               JoinWindows windows)

Join records of this stream with another KStream's records using windowed left equi join with default serializers and deserializers. In contrast to inner-join, all records from this stream will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of this KStream that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for the other stream. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of this KStream and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows), outerJoin(KStream, ValueJoiner, JoinWindows)

leftJoin

<VO,VR> KStream<K,VR> leftJoin(KStream<K,VO> otherStream,
                               ValueJoiner<? super V,? super VO,? extends VR> joiner,
                               JoinWindows windows,
                               Joined<K,V,VO> joined)

Join records of this stream with another KStream's records using windowed left equi join with default serializers and deserializers. In contrast to inner-join, all records from this stream will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of this KStream that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for the other stream. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

joined - a Joined instance that defines the serdes to be used to serialize/deserialize inputs and outputs of the joined streams

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of this KStream and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows, Joined), outerJoin(KStream, ValueJoiner, JoinWindows, Joined)

leftJoin

@Deprecated
<VO,VR> KStream<K,VR> leftJoin(KStream<K,VO> otherStream,
                               ValueJoiner<? super V,? super VO,? extends VR> joiner,
                               JoinWindows windows,
                               Serde<K> keySerde,
                               Serde<V> thisValSerde,
                               Serde<VO> otherValueSerde)

Deprecated. use leftJoin(otherStream, joiner, windows, Joined.with(keySerde, thisValSerde, otherValueSerde))

Join records of this stream with another KStream's records using windowed left equi join. In contrast to inner-join, all records from this stream will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of this KStream that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for the other stream. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(B,b)>
	<K3:c>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

keySerde - key serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

thisValSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of this KStream and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows, Joined), outerJoin(KStream, ValueJoiner, JoinWindows, Joined)

outerJoin

<VO,VR> KStream<K,VR> outerJoin(KStream<K,VO> otherStream,
                                ValueJoiner<? super V,? super VO,? extends VR> joiner,
                                JoinWindows windows)

Join records of this stream with another KStream's records using windowed outer equi join with default serializers and deserializers. In contrast to inner-join or left-join, all records from both streams will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of both KStreams that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for the this/other stream, respectively. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(null,b)> <K2:ValueJoiner(B,b)>
	<K3:c>	<K3:ValueJoiner(null,c)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of both KStream and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows), leftJoin(KStream, ValueJoiner, JoinWindows)

outerJoin

<VO,VR> KStream<K,VR> outerJoin(KStream<K,VO> otherStream,
                                ValueJoiner<? super V,? super VO,? extends VR> joiner,
                                JoinWindows windows,
                                Joined<K,V,VO> joined)

Join records of this stream with another KStream's records using windowed outer equi join with default serializers and deserializers. In contrast to inner-join or left-join, all records from both streams will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of both KStreams that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for the this/other stream, respectively. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(null,b)> <K2:ValueJoiner(B,b)>
	<K3:c>	<K3:ValueJoiner(null,c)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of both KStream and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows, Joined), leftJoin(KStream, ValueJoiner, JoinWindows, Joined)

outerJoin

@Deprecated
<VO,VR> KStream<K,VR> outerJoin(KStream<K,VO> otherStream,
                                ValueJoiner<? super V,? super VO,? extends VR> joiner,
                                JoinWindows windows,
                                Serde<K> keySerde,
                                Serde<V> thisValueSerde,
                                Serde<VO> otherValueSerde)

Deprecated. use outerJoin(otherStream, joiner, windows, Joined.with(keySerde, thisValueSerde, otherValueSerde))

Join records of this stream with another KStream's records using windowed outer equi join. In contrast to inner-join or left-join, all records from both streams will produce at least one output record (cf. below). The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key. Furthermore, two records are only joined if their timestamps are close to each other as defined by the given JoinWindows, i.e., the window defines an additional join predicate on the record timestamps.

For each pair of records meeting both join predicates the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. Furthermore, for each input record of both KStreams that does not satisfy the join predicate the provided ValueJoiner will be called with a null value for this/other stream, respectively. If an input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example (assuming all input records belong to the correct windows):

this	other	result
<K1:A>		<K1:ValueJoiner(A,null)>
<K2:B>	<K2:b>	<K2:ValueJoiner(null,b)> <K2:ValueJoiner(B,b)>
	<K3:c>	<K3:ValueJoiner(null,c)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) (for one input stream) before doing the join, using a pre-created topic with the "correct" number of partitions. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen for one or both of the joining KStreams. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Both of the joining KStreams will be materialized in local state stores with auto-generated store names. For failure and recovery each store will 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 an internally generated name, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Type Parameters:

VO - the value type of the other stream

VR - the value type of the result stream

Parameters:

otherStream - the KStream to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

windows - the specification of the JoinWindows

keySerde - key serdes for materializing both streams, if not specified the default serdes defined in the configs will be used

thisValueSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key plus one for each non-matching record of both KStreams and within the joining window intervals

See Also:

join(KStream, ValueJoiner, JoinWindows, Joined), leftJoin(KStream, ValueJoiner, JoinWindows, Joined)

join

<VT,VR> KStream<K,VR> join(KTable<K,VT> table,
                           ValueJoiner<? super V,? super VT,? extends VR> joiner)

Join records of this stream with KTable's records using non-windowed inner equi join with default serializers and deserializers. The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record that finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

See Also:

leftJoin(KTable, ValueJoiner), join(GlobalKTable, KeyValueMapper, ValueJoiner)

join

<VT,VR> KStream<K,VR> join(KTable<K,VT> table,
                           ValueJoiner<? super V,? super VT,? extends VR> joiner,
                           Joined<K,V,VT> joined)

Join records of this stream with KTable's records using non-windowed inner equi join with default serializers and deserializers. The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record that finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

joined - a Joined instance that defines the serdes to be used to serialize/deserialize inputs of the joined streams

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

See Also:

leftJoin(KTable, ValueJoiner, Joined), join(GlobalKTable, KeyValueMapper, ValueJoiner)

join

@Deprecated
<VT,VR> KStream<K,VR> join(KTable<K,VT> table,
                           ValueJoiner<? super V,? super VT,? extends VR> joiner,
                           Serde<K> keySerde,
                           Serde<V> valSerde)

Deprecated. use join(table, joiner, Joined.with(keySerde, valSerde, null))

Join records of this stream with KTable's records using non-windowed inner equi join. The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record that finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

keySerde - key serdes for materializing this (KStream input) stream If not specified the default serdes defined in the configs will be used

valSerde - value serdes for materializing this (KStream input) stream, if not specified the default serdes defined in the configs will be used

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

See Also:

leftJoin(KTable, ValueJoiner, Joined), join(GlobalKTable, KeyValueMapper, ValueJoiner)

leftJoin

<VT,VR> KStream<K,VR> leftJoin(KTable<K,VT> table,
                               ValueJoiner<? super V,? super VT,? extends VR> joiner)

Join records of this stream with KTable's records using non-windowed left equi join with default serializers and deserializers. In contrast to inner-join, all records from this stream will produce an output record (cf. below). The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record whether or not it finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. If no KTable record was found during lookup, a null value will be provided to ValueJoiner. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>			<K1:ValueJoiner(A,null)>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one output for each input KStream record

See Also:

join(KTable, ValueJoiner), leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)

leftJoin

<VT,VR> KStream<K,VR> leftJoin(KTable<K,VT> table,
                               ValueJoiner<? super V,? super VT,? extends VR> joiner,
                               Joined<K,V,VT> joined)

Join records of this stream with KTable's records using non-windowed left equi join with default serializers and deserializers. In contrast to inner-join, all records from this stream will produce an output record (cf. below). The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record whether or not it finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. If no KTable record was found during lookup, a null value will be provided to ValueJoiner. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>			<K1:ValueJoiner(A,null)>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one output for each input KStream record

See Also:

join(KTable, ValueJoiner, Joined), leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)

leftJoin

@Deprecated
<VT,VR> KStream<K,VR> leftJoin(KTable<K,VT> table,
                               ValueJoiner<? super V,? super VT,? extends VR> joiner,
                               Serde<K> keySerde,
                               Serde<V> valSerde)

Deprecated. use leftJoin(table, joiner, Joined.with(keySerde, valSerde, null))

Join records of this stream with KTable's records using non-windowed left equi join. In contrast to inner-join, all records from this stream will produce an output record (cf. below). The join is a primary key table lookup join with join attribute stream.key == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current (i.e., processing time) internal KTable state. In contrast, processing KTable input records will only update the internal KTable state and will not produce any result records.

For each KStream record whether or not it finds a corresponding record in KTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. If no KTable record was found during lookup, a null value will be provided to ValueJoiner. The key of the result record is the same as for both joining input records. If an KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream.

Example:

KStream	KTable	state	result
<K1:A>			<K1:ValueJoiner(A,null)>
	<K1:b>	<K1:b>
<K1:C>		<K1:b>	<K1:ValueJoiner(C,b)>

Both input streams (or to be more precise, their underlying source topics) need to have the same number of partitions. If this is not the case, you would need to call through(String) for this KStream before doing the join, using a pre-created topic with the same number of partitions as the given KTable. Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner); cf. join(GlobalKTable, KeyValueMapper, ValueJoiner). If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join. The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "XXX" is an internally generated name, and "-repartition" is a fixed suffix. You can retrieve all generated internal topic names via KafkaStreams.toString().

Repartitioning can happen only for this KStream but not for the provided KTable. For this case, all data of the stream will be redistributed through the repartitioning topic by writing all records to it, and rereading all records from it, such that the join input KStream is partitioned correctly on its key.

Type Parameters:

VT - the value type of the table

VR - the value type of the result stream

Parameters:

table - the KTable to be joined with this stream

joiner - a ValueJoiner that computes the join result for a pair of matching records

keySerde - key serdes for materializing this (KStream input) stream If not specified the default serdes defined in the configs will be used

valSerde - value serdes for materializing this (KStream input) stream, if not specified the default serdes defined in the configs will be used

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one output for each input KStream record

See Also:

join(KTable, ValueJoiner, Serde, Serde), leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)

join

<GK,GV,RV> KStream<K,RV> join(GlobalKTable<GK,GV> globalKTable,
                              KeyValueMapper<? super K,? super V,? extends GK> keyValueMapper,
                              ValueJoiner<? super V,? super GV,? extends RV> joiner)

Join records of this stream with GlobalKTable's records using non-windowed inner equi join. The join is a primary key table lookup join with join attribute keyValueMapper.map(stream.keyValue) == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current internal GlobalKTable state. In contrast, processing GlobalKTable input records will only update the internal GlobalKTable state and will not produce any result records.

For each KStream record that finds a corresponding record in GlobalKTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as the key of this KStream. If a KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream. If keyValueMapper returns null implying no match exists, no output record will be added to the resulting KStream.

Type Parameters:

GK - the key type of GlobalKTable

GV - the value type of the GlobalKTable

RV - the value type of the resulting KStream

Parameters:

globalKTable - the GlobalKTable to be joined with this stream

keyValueMapper - instance of KeyValueMapper used to map from the (key, value) of this stream to the key of the GlobalKTable

joiner - a ValueJoiner that computes the join result for a pair of matching records

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one output for each input KStream record

See Also:

leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)

leftJoin

<GK,GV,RV> KStream<K,RV> leftJoin(GlobalKTable<GK,GV> globalKTable,
                                  KeyValueMapper<? super K,? super V,? extends GK> keyValueMapper,
                                  ValueJoiner<? super V,? super GV,? extends RV> valueJoiner)

Join records of this stream with GlobalKTable's records using non-windowed left equi join. In contrast to inner-join, all records from this stream will produce an output record (cf. below). The join is a primary key table lookup join with join attribute keyValueMapper.map(stream.keyValue) == table.key. "Table lookup join" means, that results are only computed if KStream records are processed. This is done by performing a lookup for matching records in the current internal GlobalKTable state. In contrast, processing GlobalKTable input records will only update the internal GlobalKTable state and will not produce any result records.

For each KStream record whether or not it finds a corresponding record in GlobalKTable the provided ValueJoiner will be called to compute a value (with arbitrary type) for the result record. The key of the result record is the same as this KStream. If a KStream input record key or value is null the record will not be included in the join operation and thus no output record will be added to the resulting KStream. If keyValueMapper returns null implying no match exists, a null value will be provided to ValueJoiner. If no GlobalKTable record was found during lookup, a null value will be provided to ValueJoiner.

Type Parameters:

GK - the key type of GlobalKTable

GV - the value type of the GlobalKTable

RV - the value type of the resulting KStream

Parameters:

globalKTable - the GlobalKTable to be joined with this stream

keyValueMapper - instance of KeyValueMapper used to map from the (key, value) of this stream to the key of the GlobalKTable

valueJoiner - a ValueJoiner that computes the join result for a pair of matching records

Returns:

a KStream that contains join-records for each key and values computed by the given ValueJoiner, one output for each input KStream record

See Also:

join(GlobalKTable, KeyValueMapper, ValueJoiner)