Achieving "Fair" Event Consumption

[gremerritt]

I'm attempting to achieve "fair" event consumption across all partitions for a consumer. Let's say we have the following:

• Single broker
• Two topics test-topic-0 and test-topic-1, each with two partitions
• Single consumer subscribed to all topics (^test-topic-*)

We'll say we've already published 2 events to each partition. We don't want the consumer to process more than 1 event for a partition before moving on to others. So we'd want something like:

test-topic-0/0: 0
test-topic-0/1: 2
test-topic-1/0: 4
test-topic-1/1: 6
test-topic-0/0: 1
test-topic-0/1: 3
test-topic-1/0: 5
test-topic-1/1: 7

Based on the config docs it seemed like I'd be able to use the queued.min.messages (in this case =1) to get this behavior. However, I still see the consumer process all messages for a partition before moving to the next. Eg. something like:

test-topic-1/1: 6
test-topic-1/1: 7
test-topic-0/0: 0
test-topic-0/0: 1
test-topic-0/1: 2
test-topic-0/1: 3
test-topic-1/0: 4
test-topic-1/0: 5

I've created a simple driver for this over at https://github.com/gremerritt/kafka_fairness_test (using the ruby wrapper over librdkafka.

Also just noting I've had some success using fetch.max.bytes. Our issue is that for our usecase message sizes can vary pretty dramatically, but do not correlate well to how long processing takes, which is more strongly correlated with number of messages processed.

Any suggestions?

Edit
To put this another way - why doesn't this library implement max.poll.records as described in KIP-41? The answer I've found so far is that since this library returns a single record on poll, it's not needed. But since that setting also affects the consumers distribution of events across partitions, wouldn't it still be useful?

[saadguessous]

Hey @gremerritt, did you find any solution on how to achieve partition consumption fairness with this Kafka client?

[gremerritt]

Nope. For a bit more context on the question:

We have ~10k tenants in our system and most messages written to Kafka are scoped to a tenant. Topics are not tenant-scoped, and messages are not usually keyed by tenant, such that messages for a single tenant are evenly-ish spread across partitions. When processing messages we want tenant fairness. We currently use a home-grown solution which reads off of Kafka into queues sharded by tenant, and then we round-robin across these queues.

With KRaft allowing us to have higher partition and topic counts, our idea was to shift to a model where each topic+tenant became a topic. So where previous we'd have topic-a with messages for tenant-1 and tenant-2, we'd have two topics topic-a-tenant-1 and topic-a-tenant-2. We'd then have consumers subscribed to this set of topic-a-* topics, and rely on the client library to round-robin through them correctly. That's where this question came from.

Unfortunately, we're running on AWS MSK which caps partitions per broker at 4000 which makes the cost of this approach unfeasible. We're currently hoping to explore other potential solutions like Apache Pulsar, which may be a better fit for our usecase.

[gremerritt]

Edit: I was just re-reading the MSK docs and either I had misremembered their 4000 partition/broker limit, or they've remove that limitation. Apparently I have some reading to do :)