kafka/me.ahoo.wow.kafka

Package-level declarations

Types

class KafkaOptions @InternalWowApi constructor(val bootstrapServers: String, val properties: Map<String, String> = emptyMap(), val pollTimeout: Duration = DEFAULT_POLL_TIMEOUT, val metadataMaxAge: Duration = DEFAULT_METADATA_MAX_AGE)

How Wow reaches Kafka (kafka-clients only): the bootstrapServers, and properties passed to every producer and consumer as they are (security, client id, timeouts, batching, ...).

KafkaPartitionedSource

class KafkaPartitionedSource @InternalWowApi constructor(consumers: (String) -> Consumer<String, ByteArray>, pollTimeout: Duration, dispatcher: CoroutineDispatcher, val offsetGap: OffsetGapPolicy, observer: Observer = Observer.NONE) : PartitionedSource, AutoCloseable

The PartitionedSource over Kafka consumers: made by create from KafkaOptions, consumers without a group that commit nothing, since the runtime assigns the partitions and keeps the offsets.

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:

KafkaTopicProducer

class KafkaTopicProducer(producer: Producer<String, ByteArray>, dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher) : TopicProducer, AutoCloseable

The TopicProducer over a Kafka producer: made by create from KafkaOptions, an idempotent producer with acks=all.

KafkaTransportProvider

class KafkaTransportProvider : TransportProvider

The Kafka transport, type kafka: available once bootstrap-servers is set. Its settings, relative to its configuration (wow.kafka.*), are its options:

OffsetGapException

class OffsetGapException(val topic: String, val partition: Int, val offset: Long, val start: Long, val end: Long) : IllegalStateException

Partition partition of topic does not hold offset, the offset a read is to start from: it holds start up to end (exclusive). See OffsetGapPolicy.

OffsetGapPolicy

enum class OffsetGapPolicy : Enum<OffsetGapPolicy>

What a read of a Kafka partition does when the offset it is to read from is not in the partition: before its earliest retained offset (the topic's retention deleted what the checkpoint had yet to read), or past its end (the topic was deleted and created again, or truncated). Either way messages were lost, and the read never steps over them silently: whatever the policy, it logs an error, counts the gap (KafkaPartitionedSource.offsetGaps) and tells the node's observer of it (ObservedEvent.OffsetGap).

TopicDurability

class TopicDurability @InternalWowApi constructor(val replicationFactor: Int, val minInsyncReplicas: Int)

What a write to a topic survives: its partitions' fewest replicas (replicationFactor) and the replicas an acks=all write waits for (minInsyncReplicas, the broker's 1 when the topic sets none). A command accepted on a topic is durable once acknowledged only when both are at least 2: then losing one broker loses no acknowledged write.