wow-kafka/me.ahoo.wow.kafka/KafkaTransport

KafkaTransport

open class KafkaTransport(senderOptions: SenderOptions<String, String>, receiverOptions: ReceiverOptions<String, String>, receiverOptionsCustomizer: ReceiverOptionsCustomizer = NoOpReceiverOptionsCustomizer, receiverPolicy: KafkaReceiverPolicy = KafkaReceiverPolicy(), failurePolicy: TransportFailurePolicy = TransportFailurePolicy.DEFAULT) : Transport

Kafka as a Transport: one producer per transport, one consumer per open.

A record's key, timestamp and value are the TransportMessage's, with no headers. open subscribes the consumer group to the topics and completes readiness only after every assigned partition's position has been committed as the group's offset (anchorAssignedPartitions), so the first assignment never skips records published after readiness. Receive errors are retried with failurePolicy.

Constructors

KafkaTransport

constructor(senderOptions: SenderOptions<String, String>, receiverOptions: ReceiverOptions<String, String>, receiverOptionsCustomizer: ReceiverOptionsCustomizer = NoOpReceiverOptionsCustomizer, receiverPolicy: KafkaReceiverPolicy = KafkaReceiverPolicy(), failurePolicy: TransportFailurePolicy = TransportFailurePolicy.DEFAULT)

Types

Companion

object Companion

Properties

runtimeResource

open override val runtimeResource: RuntimeResource

The producer as a runtime resource: the runtime closes it, flushing the records it buffered, after its dispatchers stop and within the same shutdown deadline (me.ahoo.wow.runtime.RuntimeResources). Closing is idempotent, so a later close returns at once. The Kafka client cannot cancel a running close: on force stop the runtime stops waiting for it.

Functions

open override fun close()

open

open override fun open(group: String, topics: Set<String>): TransportReceiver

send

open override fun send(message: TransportMessage): Mono<Void>