kotka-streams-framework/dev.adamko.kotka.topicdata/GlobalKTableDefinition

GlobalKTableDefinition

abstract class GlobalKTableDefinition<K, V>(val tableName: String, val storeName: String = "-store", val topicName: String = "-topic", val serdes: KeyValueSerdes<K, V>, val storeType: <Error class: unknown class><K, V> = timestampedKeyValueStore()) : TopicDefinition<K, V>

An update topic topicName
plus a table-source topic tableName
plus a state store storeName

equals a GlobalTableDefinition

After defining a GlobalKTable, use buildInstance to create an instance - this has a backing topic from which the GlobalKTable will be updated.

Constructors

GlobalKTableDefinition

constructor(tableName: String, storeName: String = "-store", topicName: String = "-topic", serdes: KeyValueSerdes<K, V>, storeType: <Error class: unknown class><K, V> = timestampedKeyValueStore())

Types

data class Instance<K, V>(val updatesKStream: KStream<K, V>, val globalKTable: GlobalKTable<K, V>, val definition: GlobalKTableDefinition<K, V>)

A GlobalKTable and the backing topic from which the table is filled.

Functions

fun buildInstance(builder: StreamsBuilder): GlobalKTableDefinition.Instance<K, V>

consumeAsKStream

fun <K, V> TopicDefinition<K, V>.consumeAsKStream(builder: StreamsBuilder, consumer: Consumed<K, V> = serdes.consumer("${pid}.input-stream")): KStream<K, V>

consumeAsKTable

fun <K, V> TopicDefinition<K, V>.consumeAsKTable(builder: StreamsBuilder, consumer: Consumed<K, V> = serdes.consumer("${pid}.input-table")): KTable<K, V>

fun materializer(): Materialized<K, V, KeyValueStore<Bytes, ByteArray>>

Properties

open val pid: String

unique Kafka processor ID - to help ensure topology names are unique

open override val serdes: KeyValueSerdes<K, V>

val storeName: String

val storeType: <Error class: unknown class><K, V>

val tableName: String

override val topicName: String

The name of the backing topic, from which this table will be filled.

val topicNames: Set<String>