Ordered Queues for Kafka with Atleon

If you’ve been building asynchronous data processing systems for any meaningful amount of time, you know that Kafka is a de facto infrastructure on which such systems are commonly built. If you’ve been at it long enough, you also know about the long-tenured arguments over why “treating Kafka like a queue” (or coaxing genuine queue semantics out of it) is non-trivial. For years, that difficulty has been one of the main reasons people reach for conventional queue infrastructures instead of Kafka when what they truly (believe they) desire is a queue.

If you’ve kept up with the Kafka world lately, you’ve probably heard about KIP-932 and its delivery of so-called “Queues for Kafka.” Consumer-wise, what KIP-932 primarily introduces is a new flavor of consumer group called a share group. Share groups decouple consumers from the notion of exclusive partition ownership: Instead of being limited to one consumer per partition, you can point an arbitrary number of consumers at a topic, have them consume records from possibly-all partitions, and make use of individual queue-like acknowledgements.

Well… Not an arbitrary number of consumers. When exploring share groups, I found it notable that there does exist a meaningful limit to how many consumer instances you can have in a share group, and there is also a maximum number of outstanding record “locks” you can have per partition. These are both broker-side configurable options (as [group.share.max.size](https://kafka.apache.org/43/configuration/broker-configs/#brokerconfigs_group.share.max.size) and [group.share.partition.max.record.locks](https://kafka.apache.org/43/configuration/broker-configs/#brokerconfigs_group.share.partition.max.record.locks)), which top out at 1,000 and 10,000, respectively. These constraints are generous and I don’t think they much discount the value of share groups, but I believe these are important details to keep in mind, rather than mistakenly assume that share groups enable boundless scaling, or be surprised if you breach limits configured with these properties’ default values of 200 and 2,000.

While we’re on the subject of practical limitations, users of share groups should also be aware that records consumed via ShareConsumer have a default “lock” timeout of 30 seconds (configurable via [group.share.record.lock.duration.ms](https://kafka.apache.org/43/configuration/broker-configs/#brokerconfigs_group.share.record.lock.duration.ms)). Any polled batch of share records not acknowledged before that timeout lapses are susceptible to redelivery, and acknowledgements must be executed on all records in that batch before executing a subsequent poll, or else the consumer will implicitly consider previously polled records completed/”accepted” before returning the next batch.

Implementation nuances aside, it’s straightforward to see how “work-queue” use cases might leverage share groups. For anything else, however, the real ding in the value of share groups is you lose the guarantee of ordered consumption. In my view, ordering is one of the primary benefits, if not the selling point of using Kafka in the first place. And in all honesty, I have found true work-queue use cases to be rare; It’s almost always the case that you need (or at the very least, would prefer) to ensure that records/messages identifying the same “work” are not processed concurrently.

So I’ll be blunt: I consider share groups to be a sub-optimally-layered solution to the typically-domain-specific problems for which users may think it addresses on top of Kafka’s architecture. I somewhat suspect the appeal of share groups is driven by these problems seeming intractable when using vanilla consumer APIs, which force you to use Kafka as optimally intended. Stéphane Derosiaux excellently made the point I’m driving at here in Kafka Partitions are the Wrong Ordering Abstraction, Keys are: The partition is a storage and assignment unit (a side effect of horizontal scaling), while the key is what your domain actually cares about. It’s the same delineation between concurrency (or sharding) that is meaningful at the infrastructure level, versus ordering that is meaningful at the application level. The fact that share groups ostensibly prohibit domain-relevant processing order crystallized something for me: share groups are an attempt to solve an application-level set of problems with an infrastructure-level mechanism. It is my belief that this architectural mismatch has the potential to cause headaches if/when share groups are inappropriately leveraged.

When I pondered decomposing “treat Kafka like a queue, with support for per-key ordering” into its constituent parts, I identified three distinct problems:

1. Maintaining at least once processing in the presence of out-of-order completion (per partition).
2. Maximally mitigating head-of-line blocking when processing records from an assigned partition.
3. Conveniently facilitating per-key processing concurrency within consumption/stream processes.

The remainder of this blog will walk through each of these problems, and when paired with the right application-level machinery, show how ordered queue-like processing can be achieved with vanilla Kafka consumer groups. While I’ll be referencing machinery/code from an OSS project I maintain, Atleon, the overarching concepts are generic, and my primary goal is to prove these problems tractable rather than sell you a dependency. I won’t be disappointed, however, if you check out what else Atleon can do!

Atleon and Kafka

Problem 1: At-Least-Once Processing in the Presence of Out-of-Order Completion

The first problem to address is rooted in what Kafka fundamentally is: A log. An immutable, append-only log of records. A vanilla consumer group keeps a single committed offset per partition, which is a high-water mark indicating the point in the log up to which it has finished processing. Everything before the committed offset is done; everything at-or-after is not.

That single-offset model is wonderfully simple, and it’s exactly what gets in your way the moment you want queue-like semantics. To behave like a queue, you inherently have to tolerate a consumer processing polled records out of order, while still guaranteeing that the committed offset never advances past a record that hasn’t actually finished. Commit too eagerly, and you’ve silently dropped a message, losing the fidelity of processing at least once.

The way Atleon handles this is a mechanism called acknowledgement queuing, and the idea is simple: As each record is received (in order, before it’s ever emitted for processing), a minimal set of acknowledgement information (offset, leader epoch, etc.) is captured and appended to a linked-list-based queue. When a record finishes processing (whenever and in whatever order), its entry in the queue is marked complete. Then any contiguous range of completed entries at the head of the queue is drained and their offsets are made available for commitment.

Acknowledgement Queuing Animation

That head-anchored draining is the trick: A record that finishes early but sits behind an unfinished predecessor stays put; its offset is not eligible for commit until everything ahead of it is done. The moment the laggard at the head completes, the entire completed run behind it becomes committable at once.

Under the hood this is implemented as a single-producer, multiple-consumer (SPMC) queue, called [AcknowledgementQueue](https://github.com/atleon/atleon/blob/main/base/core/src/main/java/io/atleon/core/AcknowledgementQueue.java). Record acknowledgements are added in poll/emission order (thread-compatible, single producer), while completions arrive from processing threads in arbitrary order, with full thread safety. Each in-flight entry carries only what it needs: A “positive” acknowledger callback, a “negative” acknowledger callback, and links to its neighbors. Completion either executes acknowledgement immediately (if it’s at the head), or simply marks the node complete and lets a later drain sweep it up. The net effect is that newly committable offsets are surfaced to the underlying infrastructure as soon as it is safely possible to do so, and never a moment earlier.

Problem 2: Maximally Mitigating “Head-of-Line” Blocking

Solving Problem 1 gets us correct at-least-once fidelity, but it introduces a new failure mode, or rather, it fails to address an old one: Head-of-line blocking.

Here’s the setup: You’ve got your acknowledgement queue faithfully refusing to commit past any unfinished record. Now imagine one record at (or near) the head takes a long time to process. With a sane limit on how many pending acknowledgements you allow to be in flight, a consumer can only range that far ahead of its committable offset before it has to pause, so you eventually stall. Processing grinds to a halt, waiting on that one slow record to finish so its offset can be committed and the window can slide forward.

It’s important to point out that head-of-line blocking cannot be 100% avoided as long as core consumption tracking is high-water offset-based. However, if we break this problem into two smaller sub-problems, we can reasonably assert that it can be practically mitigated away.

Problem 2.1: Enable Polling Continuation, Even While the Head is Stuck

The first sub-problem is enabling continuation of record polling+emission, even while records ahead of the safe-to-commit point are still in flight.

Atleon addresses this with an acknowledgement queuing feature called acknowledgement compaction. The way it works is this: Suppose you have three records (in order) A, B, and C. If B and C finish processing while A is still chugging along, you don’t actually need to retain a distinct acknowledgement for B. Once A is eventually acknowledged, committing C’s offset inherently commits B at the same time, simply due to how monotonic offsets work. So B’s entry can be coalesced out of the queue the moment its subsequent neighbor completes.

Dropping B’s entry as soon as it’s redundant frees up capacity to poll and emit another record for processing. This keeps records flowing as long as you have any consecutively acknowledged records ahead of what’s safe to commit. In Atleon, acknowledgement compaction is opt-in via a queue mode: STRICT is the default, which executes every acknowledgement in order; COMPACT enables the coalescing behavior.

Problem 2.2: Durably Track Completion Ahead of Commit Offset

Acknowledgement compaction keeps you moving, but it exacerbates another issue: If your process dies/restarts or a rebalance occurs, how do you avoid reprocessing all those records that finished ahead of the committed offset? With in-memory tracking state being volatile, you need somewhere durable to record “these specific offsets ahead of the commit have already completed.”

Atleon borrows a page here from Confluent’s (now-discontinued) Parallel Consumer: Kafka natively pairs each committed offset with an optional blob of [commit metadata](https://kafka.apache.org/43/javadoc/org/apache/kafka/clients/consumer/OffsetAndMetadata.html#metadata()). That metadata is a great place to stash a set of completed-but-uncommitted offsets that sit ahead of the high-water mark. On restart/reassignment, you read that metadata back and skip emitting what’s already done.

The main catch is that this metadata blob is small/broker-bounded (configurable via [offset.metadata.max.bytes](https://kafka.apache.org/43/configuration/broker-configs/#brokerconfigs_offset.metadata.max.bytes), defaulting to 4096), so you want the encoding to be ruthlessly compact. We can apply a useful set of observations to enable optimal usage of this metadata:

1. Every completed-ahead offset is, by definition, greater than the committed offset. We can therefore represent them as unsigned deltas from the commit offset, rather than as absolute (long) values.
2. If you sort the completed offset deltas, long contiguous runs of completed offsets can further collapse under run-length encoding. A single offset writes one delta, while a long contiguous run writes a run-marker, a delta, and a count.
3. Lastly, we can make use of variable-length integer (varint) encoding so small deltas and run-lengths cost a byte or two instead of eight (since raw offsets are long values).

Applying the resulting variable-length integer based, delta run-length encoding to ahead-of-commit offset tracking causes the metadata storage size to be proportional to the number of gaps in your tracking set, and not to the number of records completed ahead of the committed offset. Internally, we can also store the completed offsets as merged ranges rather than individual longs, so the in-memory representation is gap-proportionally sized too!

Put the two halves together (acknowledgement compaction keeping polling+emission active, and durable gap-count-proportional ahead-of-commit tracking), combine them with available configuration knobs (max number of in-flight acknowledgments, and the broker-side max metadata size), and the window of completed records you can track ahead of commitment becomes controlled via configuration. Tune those configurations, which are bounded only by how much memory you’re willing to allocate, and head-of-line blocking is, for many practical purposes, mitigated away. Huzzah!

Project Reactor

Problem 3: Facilitate Per-Key Concurrency

The third problem is conveniently facilitating per-key concurrency and ordering across the partitions assigned to a consumer.

Out of the box, a vanilla Kafka consumer gives you little help here. It will happily hand you records, but if you want in-order processing of records with the same key while allowing records with different keys to be processed concurrently, that becomes an entirely application-level concern. A naive implementation — allocate per-key resources (i.e. in-memory queue, thread(s), whatever else) and route records into them — has a nasty failure mode: If you never clean those resources up, your memory/resource usage grows unboundedly as key cardinality climbs.

Before going straight to per-key concurrency, it’s worthwhile to point out that a straightforward approach to this desire for higher concurrency is to enforce a finite universe of concurrency “buckets”, which establishes a commensurate upper limit on resource allocation, reducing the need for non-trivial resource management. For example, you can group records based on hashing their keys into a fixed number of resource buckets. This is the same concept as Topic-level partitioning, except at a logical level, where cardinality can be cheaply adjusted to a concurrency level that’s much higher than the number of partitions, but not necessarily as high as the number of unique keys. Empirically, this tends to be good enough, and in many cases, most optimal.

Nevertheless, the academically-ideal level of concurrency to support is per-key, and Atleon tackles this problem by leveraging its convention of seamless positive+negative acknowledgement propagation, and its integration with Project Reactor. Since Atleon requires every emitted element to have acknowledgement executed upon processing termination, Atleon can know precisely when the last in-flight record for a given key has been processed, at which point that key’s allocated resources can be automatically disposed, rather than in some auxiliary cleanup process. Yay! No unbounded resource growth!

This “per-group automatic completion” functionality is exposed as a “group by with auto-complete” operator that plugs into Reactor’s Flux API. Each key becomes its own serialized sub-stream group, an in-flight counter tracks in-flight records per group, and the instant a group’s counter hits zero, the group is completed and removed. You get standard groupBy-style ergonomics without the sneaky problem of retaining every group for the lifetime of the (consumer-backed) source. This makes it safe for high/unbounded key cardinality, addressing where naive per-key concurrency would otherwise fall over.

Bringing it all Together

Three problems, three application-level solutions, zero infrastructure heroics:

1. Acknowledgement queueing keeps at-least-once intact even when records complete out of order.
2. Acknowledgement compaction plus gap-count-proportional, durable ahead-of-commit tracking push head-of-line blocking down to a tunable bound.
3. Auto-completing per-key grouping delivers per-key ordering with maximal concurrency and no unbounded resource growth.

Stack these all together, and you get something that has been historically difficult with regular consumer groups: queue-like consumption on top of vanilla Kafka with per-key-ordered concurrent processing, all without sacrificing at-least-once fidelity. Domain-specific concurrency lives at the application layer, where you actually understand the shape of your work, instead of contaminating your infrastructure and paying the cost of lost ordering.

The following code snippet shows what putting this all together looks like in production code:

public static void main(String[] args) throws Exception {
    // Step 1) Create config for consumer that backs receiver. This configuration includes ahead-of-commit
    //         offset tracking, acknowledgement compaction, and explicit max-in-flight control.
    KafkaConfigSource kafkaReceiverConfig = KafkaConfigSource.useClientIdAsName()
            .with(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
            .with(CommonClientConfigs.CLIENT_ID_CONFIG, "client-id")
            .with(ConsumerConfig.GROUP_ID_CONFIG, "group-id")
            .with(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
            .with(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
            .with(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
            .with(AloKafkaReceiver.OFFSET_TRACKING_STRATEGY_CONFIG, "acknowledged-ahead-of-commit")
            .with(AloKafkaReceiver.ACKNOWLEDGEMENT_QUEUE_MODE_CONFIG, AcknowledgementQueueMode.COMPACT)
            .with(AloKafkaReceiver.MAX_IN_FLIGHT_PER_SUBSCRIPTION_CONFIG, 4096);

    // Step 2) Apply stream processing to the consumed topic. The "processing" in this case
    //         introduces a superficial blocking sleep which might mimic an IO-bound process.
    Scheduler scheduler = Schedulers.newBoundedElastic(CONCURRENCY, Integer.MAX_VALUE, "example");
    AloKafkaReceiver.<String, String>create(kafkaReceiverConfig)
            .receiveAloRecords("topic")
            .groupByWithAutoComplete(ConsumerRecord::key, CONCURRENCY)
            .innerPublishOn(scheduler)
            .innerConsume(consumerRecord -> process(consumerRecord))
            .flatMapAlo()
            .doFinally(__ -> scheduler.dispose())
            .subscribe();
}

private static void process(ConsumerRecord<String, String> consumerRecord) {
    try {
        long sleepMillis = (long) (Math.random() * 10);
        System.out.printf("Processing (with sleepMillis=%d): %s", sleepMillis, consumerRecord);
        Thread.sleep(sleepMillis);
    } catch (Exception e) {
        // If desired, delegate/dead-letter errored processing here
        System.err.println("Failed to process");
    }
}

If you are on Java 25+, you can set the Reactor property reactor.schedulers.defaultBoundedElasticOnVirtualThreads to true and automatically benefit from virtual thread usage with high concurrency values.

The above example uses Atleon’s “high level” Alo API, which is where the groupByWithAutoComplete operator is available. If you happen to not need that ordering and just want unordered queue-like processing, you can also use the “low level” Reactor API:

public static void main(String[] args) throws Exception {
    // Step 1) Create options for receiver. These options includes ahead-of-commit offset tracking,
    //         acknowledgement compaction, and explicit max-in-flight control.
    KafkaReceiverOptions<String, String> receiverOptions = KafkaReceiverOptions.<String, String>newBuilder()
            .consumerProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS)
            .consumerProperty(CommonClientConfigs.CLIENT_ID_CONFIG, KafkaPerKeyParallelism.class.getSimpleName())
            .consumerProperty(ConsumerConfig.GROUP_ID_CONFIG, KafkaPerKeyParallelism.class.getSimpleName())
            .consumerProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
            .consumerProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
            .consumerProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
            .acknowledgedAheadOfCommitOffsetTracking()
            .acknowledgementQueueMode(AcknowledgementQueueMode.COMPACT)
            .maxActiveInFlight(2048)
            .build();

    // Step 2) Apply stream processing to the consumed topic. The "processing" in this case
    //         introduces a superficial blocking sleep which might mimic an IO-bound process.
    Scheduler scheduler = Schedulers.newBoundedElastic(CONCURRENCY, Integer.MAX_VALUE, "example");
    KafkaReceiver.<String, String>create(receiverOptions)
            .receiveManual(Collections.singletonList("topic"))
            .flatMap(consumerRecord -> process(consumerRecord).subscribeOn(scheduler), CONCURRENCY)
            .doFinally(__ -> scheduler.dispose())
            .subscribe();
}

private static Mono<Void> process(KafkaReceiverRecord<String, String> receiverRecord) {
    return Mono.fromRunnable(() -> {
        try {
            long sleepMillis = (long) (Math.random() * 10);
            System.out.printf("Processing (with sleepMillis=%d): %s", sleepMillis, consumerRecord);
            Thread.sleep(sleepMillis);
            receiverRecord.acknowledge();
        } catch (Exception e) {
            // If desired, delegate/dead-letter errored processing here
            System.err.println("Failed to process");
            receiverRecord.acknowledge();
        }
    });
}

You can discover more about Atleon in my previous blogs (like this one and this one), as well as on GitHub, including other novel functionalities it provides on top of Kafka.

Happy streaming!