krishvishal commented on code in PR #4205:
URL: https://github.com/apache/iggy/pull/4205#discussion_r4032967369
##########
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` |
Review Comment:
Kafka is milliseconds, `origin_timestamp` is microseconds
(`iggy_message.rs:191`), so produce multiplies by 1000 and fetch divides. Now
in the field mapping, along with the -1 case.
##########
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`
Review Comment:
Same factor both ways. The only asymmetry is an iggy-written message losing
sub-millisecond precision, since kafka has nowhere to put it.
##########
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
Review Comment:
Not our number, it is `MAX_USER_HEADERS_SIZE` (`iggy_message.rs:58`), fixed
with no knob and no rationale recorded next to it. Reworded as inherited, and
it works out to ~350 headers at the value cap, so the key-length rule is what
actually trips.
##########
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
Review Comment:
`kafka_protocol` decodes to a `Record` and keeps no raw slice, so there was
never a verbatim body to store. Layout is pinned in the doc now, 13 bytes fixed
plus 9 per header.
##########
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
Review Comment:
35 because INVALID_REQUEST is not in `maybeTransitionToErrorState`'s fatal
set, so a transactional producer goes abortable and is told to retry something
that can never work. COORDINATOR_NOT_AVAILABLE is worse, it is retriable.
##########
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.
+
Review Comment:
Kafka does not validate the sign, it stores -1 and hands it back. We cannot
(u64), so -1 maps to `delete_consumer_offset` for the same observable result,
with 3021 counted as success.
##########
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:
Done, `partition.consumer_offsets_max` is in the README with the 4096
default and the 262144 ceiling. There is no kafka code for the condition, so it
stays UNKNOWN_SERVER_ERROR with the real iggy error logged.
--
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]
