Optionally commit offsets synchronously on partition revoke

[sideeffffect]

Summary

Adds an opt-in ConsumerSettings.withCommitOnRevoke flag (default false) that, when enabled, synchronously commits the most recently requested offsets for the partitions being revoked — from inside ConsumerRebalanceListener.onPartitionsRevoked, while the consumer still owns them.

This narrows the window for duplicate record processing across consumer-group rebalances and shutdowns under at-least-once delivery.

⚠️ WIP / draft — opening for early feedback on the approach. See the open questions below.

Motivation

After a user calls .commit on a CommittableOffset, the offset is committed asynchronously (KafkaConsumer.commitAsync via the consumer actor). An async commit is only delivered/acknowledged on a subsequent poll/commitSync. If a rebalance or shutdown happens before that acknowledgement, the offset is never durably committed, the partition is reassigned, and those already-processed records are re-delivered to the new owner — duplicates.

RebalanceRevokeMode.Graceful waits for in-flight stream processing to finish before releasing partitions, but it does not, on its own, force the pending commits to land — so it doesn't fully prevent these duplicates. This is the duplicate-on-revoke behaviour discussed in #1448 (and related to #931 / the RebalanceRevokeMode work in #1399).

Approach

• New ConsumerSettings.withCommitOnRevoke(Boolean) — default false, so behaviour and overhead are unchanged unless opted in.
• When enabled, commit additionally records the latest requested offset per partition in State.requestedCommitOffsets (keeping the highest offset per TopicPartition).
• onPartitionsRevoked drains the revoked partitions' tracked offsets and commits them with commitSync before returning — i.e. while the consumer still owns them.

The tricky part: where the commit runs

Kafka invokes onPartitionsRevoked on the consumer's polling thread, from inside poll. The commitSync is therefore performed through a new internal WithConsumer.synchronouslyDuringRebalance — a direct, reentrant call on the underlying KafkaConsumer on that same thread, which Kafka explicitly permits from within a rebalance callback. Routing it through the normal single-threaded blocking context instead would submit to the executor already occupied by the in-progress poll and deadlock. (State access is bounced onto the effect runtime, since AtomicCell is thread-safe; only the consumer call itself must stay on the polling thread.)

Failures are logged and swallowed: this is an optimisation over the default async-commit behaviour, and the consumer remains at-least-once.

Limitations / open questions

• This commits only offsets the user has already requested via .commit. Records that were consumed but whose commit is still batched (e.g. commitBatchWithin), or not yet processed, are still re-delivered. It narrows the duplicate window rather than eliminating it; for stronger guarantees it should be combined with RebalanceRevokeMode.Graceful and/or idempotent processing.
• Cooperative-sticky (partial revocation): offsets are keyed by TopicPartition and only the revoked subset is committed, so it should be correct there, but I haven't exercised that path.
• API surface: is commitOnRevoke on ConsumerSettings the right place, or would you prefer this folded into RebalanceRevokeMode?
• The integration test in KafkaConsumerSpec exercises the end-to-end revoke-commit path and guards the reentrant-commit deadlock. I'm happy to strengthen it into a deterministic duplicate-vs-no-duplicate assertion if you can point me at the preferred way to induce the race in the test harness.

Note

I'm aware #1448 was closed as not planned. This PR deliberately scopes to a best-effort, opt-in mitigation of the duplicate-on-revoke case rather than attempting exactly-once or a rebalance-protocol rewrite. If you'd rather not carry it, no worries — flagging it for discussion.

🤖 Generated with Claude Code