feat(gateways): answer OffsetCommit and OffsetFetch - #4415

Closes #3542

With IGGY_KAFKA_BRIDGE_ENABLED=true, OffsetCommit (8) and OffsetFetch (9) now keep group offsets in Iggy, using the ExternalGroup kind from #4392. A group resumes from its last commit, also after a gateway restart, and kafka-console-consumer.sh --group commits and resumes.

How offsets are stored

• key: the partition, plus Iggy consumer group kafka.cg.<group> on the mapped topic. The gateway creates that group on the first commit and never joins it
• value: the Kafka offset as sent, no range check. 0 on an empty partition and one past the last message both work
• a negative offset deletes the key. Metadata and leader epoch are dropped, OffsetFetch returns "" and -1
• nothing deletes these groups on its own. OFFSET_STORAGE.md has the iggy consumer-group delete command for cleanup

Commit and fetch

• OffsetCommit v2-v9 checks the committer the way Kafka's validateOffsetCommit does. After a gateway restart a member gets UNKNOWN_MEMBER_ID and rejoins. Admin tools commit with a negative generation, which works on an empty group
• OffsetFetch v1-v9, batched groups from v8. A null topic list (Admin.listConsumerGroupOffsets) reads every topic that holds the group
• no key, no group or no topic reads as -1 with no error. Any other failed read fails the whole group instead, because a consumer that gets -1 resets its position
• retriable failures go back as 14, RequestTooOld too. Java fails offset calls on 6, and OffsetFetch on 7. A replay that already landed (RequestAlreadyApplied) counts as done
• anything sent as -1 is logged at error! with the group and the cause

Iggy calls

• offset calls run on 4 slots, and each slot has one Iggy client for commits and one for reads. Reads never wait behind a commit, and nothing waits behind Produce. The topic listing for a null topic list runs on a slot too
• a request keeps at most one call queued per slot, so one big request (a null topic list over many topics, say) can't hold up the others. Inside that, its partitions, the groups of a v8 fetch and the topics of a null topic list run at once
• each partition sticks to one slot, and a topic's partitions take the slots in turn. A commit holds its slot until it ends, 45s at most. Commits on a slot share one client that sends one call at a time, which keeps them in order even when a caller gives up. A read frees its slot at the deadline
• a request gives up after half the min session timeout (3s by default). Heartbeats on the same connection wait behind it, so a request also gives up after half of what's left of the session of the member that heartbeats there. Partitions cut off by that answer 14 without a call

Server change

get_consumer_offset answered a read that timed out on the owning shard, or found the partition not loaded there, with an empty body. Over HTTP the not-loaded case was a 404. The SDK reads both as "no offset", and a consumer would reset. Both now answer TransientNotAccepted (503 over HTTP), which the Rust SDK retries. This is in core/server/src/dispatch/partition.rs and core/server/src/http/handlers.rs and applies to every client.

Known limits

• every Iggy server needs a build with kind 3 (feat(server): store offsets of groups managed outside Iggy #4392) before the first commit. Nothing enforces that yet, and in a partly upgraded cluster one commit can stop a partition from committing
• in a cluster an offset read can land on a node that lags. Right after a group's first commit that node answers -1, and a consumer on auto.offset.reset=latest skips records. Reads from the primary are Serve consumer offset reads from the partition primary #4409
• with more than one gateway, a commit with no generation on a gateway without the group's members overwrites the live offsets, and a commit a gateway gave up on can land after a newer one from another gateway. Both are in OFFSET_STORAGE.md
• delivery stays at-least-once