kafka/me.ahoo.wow.kafka/KafkaTopicAdmin

KafkaTopicAdmin

class KafkaTopicAdmin(admin: Admin, replicationFactor: Short? = null, configs: Map<String, String> = emptyMap(), visibleWithin: Duration = DEFAULT_VISIBLE_WITHIN, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher) : TopicAdmin, AutoCloseable

The TopicAdmin over a Kafka admin client, made by create from KafkaOptions:

partitions describes the topics, a topic the broker does not know left out; it never creates one.
create creates the missing topics with replicationFactor (the broker's default.replication.factor when null) and the topic configs (retention.ms, min.insync.replicas, ...); a topic that exists already is left as it is. It then waits, at most visibleWithin, until every topic can be described, as metadata reaches the brokers a moment after the controller created it.
durability reads what a topic's writes survive: its replication factor and its min.insync.replicas, which an acks=all producer's acknowledgement rests on.

The admin's blocking futures are awaited on virtual threads; Kafka being unreachable, or refusing the node's credentials, fails a call as a me.ahoo.wow.spi.transport.TransportUnavailableException (KafkaFailures). close closes the admin client.

Constructors

KafkaTopicAdmin

constructor(admin: Admin, replicationFactor: Short? = null, configs: Map<String, String> = emptyMap(), visibleWithin: Duration = DEFAULT_VISIBLE_WITHIN, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher)

Types

Companion

object Companion

Functions

open override fun close()

create

open suspend override fun create(topics: Map<String, Int>)

durability

suspend fun durability(topics: Set<String>): Map<String, TopicDurability>

The replication factor and min.insync.replicas of each of topics the broker knows: what an acks=all write to it survives. A topic the broker does not know is left out.

partitions

open suspend override fun partitions(topics: Set<String>): Map<String, Int>