Kafka: a fully-drained consumer group is left permanently one offset short per partition, redelivering on every rebalance

## Summary

After a Kafka consumer has handled every message on its partitions, the committed offset settles **one short of the log end on every partition** and stays there. `kafka-consumer-groups.sh --describe` reports `LAG 1` per partition indefinitely for a group that is fully caught up, and every subsequent rebalance redelivers one message per partition.

Found while writing a tutorial sample (#4280), not in production. Not urgent, and I have deliberately **not** asserted a root cause — see *What I could not explain* below.

## Reproduce

`samples/Tutorials/04-Kafka` from #4280, or any `KafkaSubscription` at defaults. Three partitions, three messages each, `commitBatchSize` left at its default of 10 so the batch path never fires.

Immediately after all nine messages are handled, nothing is committed at all — expected, since the batch was never reached:

```text
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
greeting.event 0 - 3 -
greeting.event 1 - 3 -
greeting.event 2 - 3 -
```

The 30-second `sweepUncommittedOffsetsInterval` timer then fires, and the group settles at:

```text
greeting.event 0 2 3 1
greeting.event 1 2 3 1
greeting.event 2 2 3 1
```

**That is the end state, not a transient.** Sampled at t+15s, t+30s, t+45s and t+60s — spanning at least one further sweep — all four identical. Reproduced across four runs on torn-down brokers.

## What the consumer's own log shows

All three offsets per partition are stored *before* the sweep runs, and the sweep lists all nine:

```text
STORE offset 1 ... partition 2 STORE offset 1 ... partition 0 STORE offset 1 ... partition 1
STORE offset 2 ... partition 2 STORE offset 2 ... partition 0 STORE offset 2 ... partition 1
STORE offset 3 ... partition 2 STORE offset 3 ... partition 0 STORE offset 3 ... partition 1

Sweeping offsets:
Topic: greeting.event Partition: 2 Offset: 1
Topic: greeting.event Partition: 2 Offset: 2
Topic: greeting.event Partition: 2 Offset: 3
Topic: greeting.event Partition: 0 Offset: 1
... (same shape for partitions 0 and 1)
```

So the highest stored offset for each partition is **3**, it is present in the list handed to `Commit`, and the committed result is **2**.

## The contributing factor I am confident about

`_offsetStorage` is a `ConcurrentBag` (`KafkaMessageConsumer.cs:54`). `CommitAllOffsets` drains it and passes the whole list to `_consumer.Commit(...)` (`KafkaMessageConsumer.cs:958-982`) **without reducing to one entry per `TopicPartition`**. A list containing three different offsets for the same partition is being handed to a call whose behaviour on duplicates is not something the calling code should be relying on.

Whatever librdkafka does with those duplicates, passing them looks unintended, and reducing to `max(offset)` per `TopicPartition` before committing would make the outcome deterministic and correct.

## What I could not explain, and did not want to guess at

**Why the committed offset is 2 rather than 1 or 3.** On the run logged above the list is in *ascending* order per partition, so neither "first entry wins" (1) nor "last entry wins" (3) accounts for it. A second run drained the bag in *descending* order and also settled at 2. I have not instrumented librdkafka to find out, and I would rather report the measurement than publish a mechanism I inferred.

Two reviewers looking at the same evidence each proposed a different mechanism — "last write wins" and "the sweeper is a race that resolves itself" — and the logs above are consistent with neither, which is why this section exists.

## Why it is worth fixing regardless

- A healthy, fully-drained consumer group **never reports `LAG 0`**, so lag-based alerting and dashboards are permanently misleading by one message per partition.
- **Every rebalance redelivers.** Measured on the same sample: a second instance joining after the sweep causes 3 of 9 messages to be handled twice, one per partition. At-least-once permits this, but it is redelivery that no message-handling failure caused and that no amount of waiting clears.
- The narrower the batch relative to traffic, the more visible it is — a low-volume topic sits at `LAG 1` forever.

## Suggested fix

In `CommitAllOffsets` (and `CommitOffsetsFor`, which has the same shape), group the drained offsets by `TopicPartition` and commit only the maximum offset for each. That is a small change, makes the committed position independent of bag ordering, and should take the steady state to `LAG 0`.

## Related but not the same

- #3586 — offsets not committed on shutdown, and `Skipped committing offsets, as another commit or sweep was running`. That is a contention/shutdown issue where *nothing* commits; here the commit happens and lands low.
- #3055 — spurious `StoreOffset` call.

## Environment

Brighter `master` at `10351e970`, `net9.0`, `apache/kafka:4.0.2` (KRaft, single broker) from `docker-compose-kafka.yaml`, Confluent .NET client, all `KafkaSubscription` defaults except `numOfPartitions: 3`.