KafkaPartitionedSource

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.

• partitions and offsets ask one consumer of their own (consumers makes it on the first call), one call at a time, since a Kafka consumer is not safe for concurrent use.

• partitionOf is the producer's default partitioner: murmur2 of the key over the partition count.

• read of a partition the source does not read yet joins the topic's shared consumer (SharedTopicConsumer): the tails of every partition of a topic this node owns go through one consumer, assigned those partitions by hand. A second read of a partition already joined, such as a subscription catching up, and a bounded read run a consumer of their own for as long as their flow is collected: assigned the one partition, it seeks to from and polls, each poll at most pollTimeout on a virtual thread, emitting what it got; the consumer closes when the flow ends or is cancelled.

Either way, an offset the partition does not hold, before its earliest retained offset or past its end, is a gap: the read never steps over it silently, but logs an error, counts it in offsetGaps and does what offsetGap says, by default failing with an OffsetGapException.

Each consumer is named after what it does (consumers takes the name, client.id with KafkaOptions): metadata for partitions and offsets, tail.<topic> for a topic's shared consumer and read.<topic>-<partition> for a read of its own, each with a number unique in the process, so an operator tells them apart in the client's logs and metrics, and the client registers each once.

close closes the consumer of partitions and offsets and the shared consumers; the other reads close their own.