ryerraguntla commented on code in PR #4205: URL: https://github.com/apache/iggy/pull/4205#discussion_r4031403663
########## gateways/kafka/docs/OFFSET_STORAGE.md: ########## @@ -0,0 +1,132 @@ +# Consumer offset storage + +Status: proposed. Answers [#3540](https://github.com/apache/iggy/issues/3540) and blocks +[#3542](https://github.com/apache/iggy/issues/3542), OffsetCommit and OffsetFetch. + +## Decision + +Store Kafka group offsets as Iggy consumer offsets, one key per partition, under a consumer +group whose name is derived from the Kafka group id. + +The issue lists three options. None of them is this one. + +| Option | Why not | +| -------- | --------- | +| A, an Iggy-backed `__consumer_offsets` topic | Rebuilds what Iggy already has. A compacted offset topic needs compaction, which Iggy does not have, so the gateway replays the whole topic at every startup | +| B, a SQLite file on the gateway host | A second durability story, a second backup story, and offsets that do not survive moving the gateway | +| C, in memory only | Fails the acceptance criterion in #3542, which is that offsets survive a restart | + +Iggy already stores a durable offset per consumer and per partition, replicated with the +partition itself. Using it costs one call per partition on commit and one on fetch. + +## The key + +One Iggy consumer offset per Kafka `(group, topic, partition)`. + +- consumer kind: `ConsumerKind::ConsumerGroup` +- consumer id: `Identifier::named("kafka.cg.<group>")` +- stream and topic: whatever `TopicMapping` resolves the Kafka topic to +- partition: the Kafka partition index, unchanged, because both sides number from 0 + +The gateway calls `create_consumer_group(stream, topic, "kafka.cg.<group>")` before the first +commit for a group on a topic. If the group does not resolve in metadata, the server rejects the +offset write. The group has to exist first. The gateway never joins the group. Offsets are +readable by any client, member or not. + +### Why the group kind and not a named consumer + +`ConsumerKind::Consumer` with a name looks simpler, because it needs no registration call. It is +not. The server hashes a named consumer id to a `u32` with `XxHash32` +(`core/server/src/dispatch/partition.rs:916`), and that hash is the offset key. Two different +group names can collide and silently share one offset. + +A consumer group name resolves through metadata to a monotonic id instead. No hash, no +collision, and `get_consumer_groups(stream, topic)` lists what exists. + +### Why the prefix + +`kafka.cg.` keeps a Kafka group called `orders` off the key that a native Iggy consumer group +called `orders` uses. Without it the two share an offset and each one moves the other. + +The prefix does not make the offsets safe to poll with. That is the next section. + +## What is stored + +The Kafka committed offset, verbatim, with no conversion. + +The two systems mean different things by the number. Kafka commits the next offset to read. +Iggy stores the last offset processed, and `PollingKind::Next` resumes at the stored value plus +one (`core/partitions/src/iggy_partition.rs:3835`). A Kafka offset stored in an Iggy key is +therefore one greater than Iggy's own convention for that key. + +This is inert because the gateway never polls that way. Fetch always polls with an explicit +offset, `PollingKind::Offset`, taken from the Kafka request. Nothing in the gateway reads the +stored value to decide where to resume. It is returned to the client on OffsetFetch and +otherwise untouched. + +The rule this creates: no code path polls a `kafka.cg.*` key with `PollingKind::Next`. Doing so +skips one record per partition. The prefix is what keeps a native Iggy consumer from doing it by +accident. + +Converting on write instead, and storing the Kafka offset minus one, breaks at offset 0. It also +makes an empty commit look the same as a commit of the first record. Storing verbatim is the +smaller problem. + +## OffsetFetch with no topics named + +OffsetFetch v2 and later let a client pass a null topic list, which asks for every offset the +group holds. `kafka-consumer-groups.sh --describe` does this. + +Iggy has no lookup by consumer. Offsets are read one partition at a time +(`core/common/src/traits/consumer_offset_client.rs:41`). The gateway answers by enumerating the +topics in the mapped stream and querying each partition of each one. + +That is one round trip per partition on an admin call. The cost is bounded by the topic and +partition count of one stream. This is an admin path and not a data path, so the cost is +acceptable. It is written here so nobody discovers it in a test. + +## What is dropped + +Kafka lets a client attach a metadata string to a commit. Iggy stores a number and nothing else. +The string is dropped on commit, and OffsetFetch returns an empty string. + +`committed_leader_epoch` is dropped the same way. The gateway reports `-1`. + +## Limits + +A partition admits a bounded number of distinct offset keys per consumer kind. The default is Review Comment: 4096 offset keys/partition/kind → Kafka -1 UNKNOWN_SERVER_ERROR. Multi-tenant group count will hit this with no useful client retry. Call the operator knob (partition.consumer_offsets_max) in README -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
