This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-bridge-mapping in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ffbbed3ca2dd1d840dae4fcbc1d44dc0506797ed Author: Krishna Vishal <[email protected]> AuthorDate: Wed Sep 16 23:58:03 2026 +0530 docs(gateways): settle the Kafka record mapping, producer ids, offsets Produce (#3535) and Fetch (#3536) cannot be designed until these are written down. --- gateways/kafka/README.md | 6 + gateways/kafka/docs/BRIDGE_MAPPING.md | 204 ++++++++++++++++++++++++++++++++++ gateways/kafka/docs/IDEMPOTENCE.md | 92 +++++++++++++++ gateways/kafka/docs/OFFSET_STORAGE.md | 132 ++++++++++++++++++++++ gateways/kafka/docs/SCOPE.md | 12 +- 5 files changed, 443 insertions(+), 3 deletions(-) diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index a1e10c010..23698538b 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -58,6 +58,12 @@ Before check-in, run the procedure in [docs/MANUAL_TESTING.md](docs/MANUAL_TESTI See [docs/SCOPE.md](docs/SCOPE.md) for [#3421](https://github.com/apache/iggy/issues/3421) deliverables, supported API key/version table, and post-foundation TODO backlog. +## Design decisions + +- [docs/BRIDGE_MAPPING.md](docs/BRIDGE_MAPPING.md) — how a Kafka record becomes an Iggy message, and back +- [docs/IDEMPOTENCE.md](docs/IDEMPOTENCE.md) — InitProducerId, and why delivery is at-least-once +- [docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md) — where Kafka consumer group offsets live + ## Iggy bridge ([#3533](https://github.com/apache/iggy/issues/3533)) `src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy diff --git a/gateways/kafka/docs/BRIDGE_MAPPING.md b/gateways/kafka/docs/BRIDGE_MAPPING.md new file mode 100644 index 000000000..86cf53c85 --- /dev/null +++ b/gateways/kafka/docs/BRIDGE_MAPPING.md @@ -0,0 +1,204 @@ +# Kafka to Iggy record mapping + +Status: proposed. Direction agreed with @spetz and @hubcio on 2026-09-16 ("messages should be +stored in iggy format"); the escape hatches below still need sign-off. Closes the last open +scope item of [#3533](https://github.com/apache/iggy/issues/3533) and blocks +[#3535](https://github.com/apache/iggy/issues/3535) (Produce) and +[#3536](https://github.com/apache/iggy/issues/3536) (Fetch). + +## Decision + +One Kafka record becomes one Iggy message, in Iggy's own format: the record value is the +message payload, the key and the Kafka headers become Iggy user headers. The gateway rebuilds +a Kafka record batch on Fetch. + +Two properties drive this. + +A Kafka consumer must be able to read a topic an Iggy producer wrote. This is the staged +migration the maintainers described: rewrite producers to the Iggy SDK first, leave consumers +on the gateway until later. The gateway can only encode an arbitrary Iggy message as a Kafka +record if the stored form has no Kafka framing in it. + +Kafka offsets must line up with Iggy offsets. A Kafka record batch carries N records under one +base offset, while an Iggy message consumes exactly one offset. Storing a batch whole makes +every offset the gateway reports wrong by the batch size, and fixing that inside Iggy means +teaching the server to count records inside an opaque payload. + +Storing the Kafka payload and headers as a dump inside the Iggy payload was the alternative +raised in the same thread. It does not satisfy the first property on its own: the gateway would +still need a native path for Iggy-written messages, so it would carry two storage formats +instead of one. Native storage with a narrow fallback (see below) keeps that to one. + +## Field mapping + +Produce, per record: + +| Kafka | Iggy | +| ------- | ------ | +| record value | `payload` | +| record key | `kafka.key` user header, `Raw` | +| record header `name` | `kafka.h.<name>` user header, `Raw` | +| record timestamp (CreateTime) | `origin_timestamp` | +| record offset | partition offset, assigned by Iggy | +| partition index | partition index, both 0-based | +| topic | stream and topic per `TopicMapping` | + +Fetch reverses it. A message with no `kafka.*` headers is a message an Iggy client wrote, and +encodes as a Kafka record with a null key, its Iggy user headers as Kafka headers, and +`origin_timestamp` as the record timestamp (falling back to the server-assigned `timestamp` +when the origin timestamp is zero). Iggy header kinds other than `Raw` and `String` are emitted +as their raw value bytes. + +## Records Iggy cannot hold natively + +Iggy rejects an empty payload (`core/common/src/types/message/iggy_message.rs:169`), caps a +user header value at 255 bytes (`core/common/src/types/message/user_headers.rs:631`), keys +headers in a `BTreeMap` so a name cannot repeat, and caps all user headers of a message at +100 KB (`MAX_USER_HEADERS_SIZE`). Kafka allows all of the shapes those rules exclude, so two +mechanisms cover them. + +### Null and empty values + +A record with a null value (a tombstone) or a zero-length value is stored with a single `0x00` +byte payload and a `kafka.value` header holding `null` or `empty`. Fetch reads that header and +restores the original, discarding the placeholder byte. + +This keeps tombstones on the fast path rather than pushing them into the fallback, because they +are ordinary traffic on compacted Kafka topics. Iggy has no compaction, so a tombstone is stored +and served like any other record and nothing acts on it. + +### Everything else: the envelope fallback + +A record takes the fallback when any of these hold: + +- the key is longer than 255 bytes +- a header name, prefixed with `kafka.h.`, is longer than 255 bytes +- a header value is null, empty, or longer than 255 bytes +- two headers share a name +- the headers together would exceed the 100 KB user-header budget + +Such a record is stored with a `kafka.envelope` header carrying the format version, and the +Kafka record body (key, value, headers) verbatim in the payload. Fetch checks for that header +first and takes the plain path only when it is absent. + +The cost is that these messages are opaque to Iggy consumers and connectors. That is the point +of confining the fallback to record shapes that are rare in practice, rather than making it the +default storage form. + +## Batch-level fields + +Per-record storage drops what the Kafka record batch header carries: producer id, producer +epoch, base sequence, the transactional flag, compression and the batch CRC. Fetch synthesizes +a batch with producer id `-1`, epoch `-1`, base sequence `-1`, no compression, `CreateTime` +timestamps, and a recomputed CRC32C. + +Two consequences worth stating before they surprise someone: + +- Idempotent-producer deduplication cannot be reconstructed from stored data later. If + [#3545](https://github.com/apache/iggy/issues/3545) ever grows past a stub, producer id, + epoch and sequence need their own tracking. +- The bytes a consumer receives are not the bytes the producer sent, so anything comparing + batches byte for byte across the gateway will differ. + +Whether producer id `-1` is what Fetch actually sends depends on the InitProducerId decision in +[`IDEMPOTENCE.md`](IDEMPOTENCE.md). Allocating producer ids does not change what is stored, only +what Produce accepts, so this section holds under either answer. + +Produce decompresses gzip, snappy, lz4 and zstd batches, which means turning those features back +on for the `kafka-protocol` dependency (`Cargo.toml:217` currently builds it with +`default-features = false, features = ["broker"]`). Fetch emits uncompressed batches. + +## Offsets + +Kafka offset and Iggy offset are the same number for the same record, and both partition spaces +are 0-based, so neither direction converts. + +Produce takes the base offset from the send confirmation +(`SendMessagesConfirmationResponse::base_offset`). The server may return no confirmation, for +example for a request it classifies as a duplicate, in which case the response carries `-1` +rather than a guessed offset; Kafka clients surface that as an unknown offset. + +ListOffsets LATEST is the high watermark from `IggyBridge::high_watermarks`. EARLIEST has no +server-side field today (`Partition` carries no log start offset), so it reads the first +retained message instead, and the `(messages_count, current_offset) == (0, 0)` ambiguity +documented on `high_watermarks` applies to both. + +## Partitioning + +Both systems number partitions from 0, so the partition index passes through unchanged in each +direction and neither side converts. + +Produce sends to the partition the request names, `Partitioning::partition_id(index)`. A Kafka +producer resolves the partition itself before it builds the request, so every partition index in +a `ProduceRequest` is a real one and `Partitioning::balanced()` has no trigger on this path. The +`-1` that `SCOPE.md` refers to belongs to CreateTopics, where it means "use the broker default +partition count", and it is handled there rather than here. + +Kafka consumer groups are not mapped onto Iggy consumer groups. The gateway assigns partitions +to group members the way Kafka does, in the client, and polls every partition by explicit offset. +Iggy's group registry is used as an offset key and for nothing else, which +[`OFFSET_STORAGE.md`](OFFSET_STORAGE.md) covers. + +## Reserved header namespace + +`kafka.` is reserved on messages the gateway writes and reads. An Iggy producer that sets a +header in that namespace on a topic a Kafka consumer reads will have it interpreted as gateway +metadata. + +## Open questions + +Four questions need an answer before Produce ([#3535](https://github.com/apache/iggy/issues/3535)) +is written. Each one carries a default. If no answer lands by 2026-09-22, the default is taken, +this document is updated to record that it was decided by default, and the work proceeds. + +### 1. Envelope fallback, or reject the record? + +A record that Iggy cannot hold natively goes into the envelope described above. The alternative +is to reject it with `MESSAGE_TOO_LARGE` (10), so that nothing an Iggy consumer cannot read ever +reaches a stream. + +Rejecting is the stricter guarantee and the worse compatibility story: a Kafka producer that +sends a 300-byte key works against a real broker and fails against the gateway. + +Default: keep the envelope. + +### 2. Is `kafka.` the right prefix? + +Every Kafka header name is stored as `kafka.h.<name>`, which spends 8 of the 255 bytes an Iggy +header name has, on every header of every record. A shorter prefix buys those bytes back and +costs readability for anyone reading a stream by hand. + +Default: keep `kafka.`. + +### 3. Is the placeholder byte acceptable for tombstones? + +A null or empty value is stored as one `0x00` byte plus a `kafka.value` marker header. The +payload a native Iggy consumer sees is therefore a byte the producer never sent. + +The alternative is the envelope, which costs a tombstone the fast path. Tombstones are ordinary +traffic on compacted Kafka topics, and Iggy has no compaction, so they are stored and served +like any other record. + +Default: keep the placeholder byte. + +### 4. Recompress on Fetch, or always emit uncompressed? + +Produce decompresses, and Fetch currently rebuilds an uncompressed batch. Recompressing per +topic costs CPU on the read path and saves bytes on the wire to the consumer. + +Default: always emit uncompressed, and revisit when a benchmark says it matters. + +### Not asked here + +A Kafka `retention.ms` topic config could map onto Iggy's `message_expiry` at creation time. +`ensure_stream_and_topic` leaves topics on the server default, which never expires. That belongs +to CreateTopics ([#3538](https://github.com/apache/iggy/issues/3538)), which owns topic +configuration, rather than to the record mapping. + +## References + +- Scope and phases: [`SCOPE.md`](SCOPE.md) +- Bridge API: `gateways/kafka/src/bridge/iggy_bridge.rs` +- Iggy message limits: `core/common/src/types/message/iggy_message.rs`, + `core/common/src/types/message/user_headers.rs` +- Produce confirmations: `core/binary_protocol/src/responses/messages/send_messages.rs` diff --git a/gateways/kafka/docs/IDEMPOTENCE.md b/gateways/kafka/docs/IDEMPOTENCE.md new file mode 100644 index 000000000..a3ea1f093 --- /dev/null +++ b/gateways/kafka/docs/IDEMPOTENCE.md @@ -0,0 +1,92 @@ +# InitProducerId and idempotent producers + +Status: proposed. Answers the open half of +[#3545](https://github.com/apache/iggy/issues/3545) and gates the Phase 1 end-to-end test +([#3539](https://github.com/apache/iggy/issues/3539)), which drives +`kafka-console-producer.sh`. + +## The problem + +A stock Java producer sets `enable.idempotence=true` without being asked. That default arrived +in Kafka 3.0 and took effect from 3.0.1, 3.1.1 and 3.2.0, where a bug that suppressed it was +fixed. `kafka-console-producer.sh` leaves it on. + +An idempotent producer sends InitProducerId (key 22) before its first record. The gateway does +not list key 22, so ApiVersions does not advertise it, and the producer raises +`UnsupportedVersionException`. That exception is fatal. +`TransactionManager.maybeTransitionToErrorState` tests it above the `isTransactional()` branch. +The producer therefore enters a fatal error state instead of dropping back to weaker semantics. +It fails at startup, before it sends a record. + +The gateway's stated purpose is that a Kafka user swaps the broker and changes no application +code. A broker that the default producer cannot start against does not meet it. + +## Options + +| Option | Cost | What a stock producer does | +| -------- | ------ | ---------------------------- | +| Stub with `UNSUPPORTED_VERSION` | none | fails at startup unless the user sets `enable.idempotence=false` | +| Allocate only | about a day | works untouched, at-least-once delivery | +| Real deduplication | large | works untouched, exactly once per partition, single gateway instance only | + +Real deduplication means tracking a sequence number per producer and per partition, and +rejecting a duplicate or a gap. It is correct only while one gateway instance sees every write +from a producer, so it cannot be decided before the multi-instance question is. + +## Decision + +Allocate only. + +Rejecting the stock producer to avoid implementing deduplication trades the one requirement the +maintainers named against a guarantee Iggy does not offer today anyway. Allocating costs about +a day and keeps delivery exactly where it already is. + +## Behavior + +Add key 22 to `SUPPORTED_RANGES` in `src/protocol/api.rs` and advertise it through ApiVersions. +Without both, the producer never sends the request. `kafka-protocol` 0.18 carries the schemas, +request v0 to v5 and response v0 to v6, flexible from v2. + +InitProducerId with no `transactional_id`: + +- allocate the next producer id, return it with epoch 0 and error code 0 +- draw ids from a counter seeded per gateway instance, so two instances never hand out the same + id. Nothing reads the id today. Seeding it now is what stops a later deduplication layer from + being born broken + +InitProducerId with a `transactional_id`: + +- answer `UNSUPPORTED_VERSION` (35), unchanged. Transactions stay out of scope, and so do + AddPartitionsToTxn (24), AddOffsetsToTxn (25), EndTxn (26) and TxnOffsetCommit (28) + +Produce: + +- accept `producer_id`, `producer_epoch` and `base_sequence` on the request and ignore them +- never answer `OUT_OF_ORDER_SEQUENCE_NUMBER` (45) or `DUPLICATE_SEQUENCE_NUMBER` (46). The + gateway tracks no sequences, so it cannot tell the two apart from ordinary traffic + +## What this does not give you + +A producer that has an id believes its retries are deduplicated. They are not. A retry after a +network timeout writes the record twice, and both copies reach the stream with their own +offsets. + +Delivery through the gateway is at-least-once, with or without this change. The difference is +that the producer now starts. + +State that limitation in the README, next to the transaction section, in those words. Do not +leave a user to infer it from the presence of key 22. + +## Open question + +Allocate only, as above, or stub and document `enable.idempotence=false`? + +If no answer lands by 2026-09-22, allocate only is taken and the work proceeds. This document +is then updated to record that it was decided by default. + +## References + +- Record mapping: [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md), batch-level fields +- Scope and phases: [`SCOPE.md`](SCOPE.md) +- Version firewall: `src/protocol/api.rs`, `SUPPORTED_RANGES` +- Fatal path: `TransactionManager.maybeTransitionToErrorState`, apache/kafka trunk diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md new file mode 100644 index 000000000..4f9a53bcf --- /dev/null +++ b/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 +4096, set by `partition.consumer_offsets_max` +(`core/configs/src/server_config/partition.rs:151`). The configurable ceiling is 262144. Passing +the limit returns `TooManyConsumerOffsets` (3024). Kafka has no error code for this condition, so +it maps to `UNKNOWN_SERVER_ERROR`, which is what `bridge/error.rs` already does where no honest +code exists. + +Consumer groups and plain consumers count against separate limits, so Kafka groups do not +compete with native Iggy consumers for the same 4096. + +An Iggy name is capped at 255 bytes (`core/common/src/lib.rs:168`), which leaves 246 for a Kafka +group id after the prefix. A longer group id is rejected with `INVALID_GROUP_ID` (24). + +## More than one gateway instance + +Two gateway instances that share an Iggy cluster read and write the same offset keys. The key +comes from the Kafka group id and nothing else. Two instances that serve one group therefore +agree on committed offsets without talking to each other. + +They do not agree on group membership. That belongs to the coordinator +([#3541](https://github.com/apache/iggy/issues/3541)) and is not settled here. + +## Open question + +Iggy consumer offsets keyed by group, as above, or one of A, B and C from the issue? + +If no answer lands by 2026-09-22, the design above is taken and the work proceeds. This document +is then updated to record that it was decided by default. + +## References + +- Record mapping: [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md) +- Scope and phases: [`SCOPE.md`](SCOPE.md) +- Offset API: `core/common/src/traits/consumer_offset_client.rs` +- Group API: `core/common/src/traits/consumer_group_client.rs` +- Offset key resolution: `core/server/src/dispatch/partition.rs` diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index de7a343de..30289a934 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -109,10 +109,10 @@ below it are still open for the issues that build on top of it. not part of `bridge/`'s own scope. - [x] Idempotent `ensure_stream_and_topic()` (create-if-not-exists) - `src/bridge/iggy_bridge.rs`, exercised end-to-end in `tests/bridge_iggy_integration_tests.rs`. -- [ ] Document partition mapping in `docs/BRIDGE_MAPPING.md`: +- [x] Document partition mapping in [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md): - Iggy partitions are **0-based** (same as Kafka) — direct `partition_id` mapping, no offset conversion - - Iggy **consumer groups exist** — map Kafka group APIs to Iggy consumer group APIs - - Use `Partitioning::balanced()` only when Kafka sends `partition == -1`; otherwise use request partition ID + - Kafka consumer groups do **not** map onto Iggy consumer groups. Assignment stays client-side, and Iggy's group registry is used as an offset key only ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)) + - `Partitioning::partition_id(index)` on every Produce. A Kafka producer resolves the partition before it builds the request, so `Partitioning::balanced()` has no trigger there. The `-1` default-partition-count case belongs to CreateTopics - [ ] Real Metadata topology (brokers, partitions, leaders) backed by Iggy state ### `kafka-protocol` crate adoption — superseded, done differently @@ -130,12 +130,18 @@ above). ### Phase 3 — Consumer groups (~7 API keys) +Offset persistence design ([#3540](https://github.com/apache/iggy/issues/3540)): +[`OFFSET_STORAGE.md`](OFFSET_STORAGE.md). + - [ ] OffsetCommit (8), OffsetFetch (9), FindCoordinator (10) - [ ] JoinGroup (11), Heartbeat (12), LeaveGroup (13), SyncGroup (14) - [ ] DescribeGroups (15), ListGroups (16) as needed by target clients ### Phase 3+ — Auth, admin, tuning +InitProducerId and idempotent producers +([#3545](https://github.com/apache/iggy/issues/3545)): [`IDEMPOTENCE.md`](IDEMPOTENCE.md). + - [ ] SASL (17, 36) if required by deployment - [ ] Tune `max_frame_size` per workload (Kafka defaults: ~1 MiB produce, ~50 MiB fetch; current default 8 MiB) - [ ] Target **~15–20 API keys** total for a functional bridge — not all 74+ admin keys
