You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/09/24 08:03:34 UTC
(iggy) 01/03: feat(gateways): send Kafka records to Iggy from the Produce handler
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-produce
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8e5364112938219517ba150ff7bd70262df5e357
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Sep 22 18:03:52 2026 +0530
feat(gateways): send Kafka records to Iggy from the Produce handler
The record codec and the bridge were both in place, so Produce was the only thing still discarding every payload it decoded.
---
.config/nextest.toml | 10 +-
gateways/kafka/README.md | 78 ++-
gateways/kafka/docs/BRIDGE_MAPPING.md | 8 +-
gateways/kafka/docs/MANUAL_TESTING.md | 19 +-
gateways/kafka/docs/SCOPE.md | 15 +-
gateways/kafka/docs/TEST_SUITE.md | 9 +-
gateways/kafka/src/bridge/error.rs | 17 +-
gateways/kafka/src/bridge/iggy_bridge/produce.rs | 62 ++-
gateways/kafka/src/protocol/api.rs | 12 +
gateways/kafka/src/protocol/handlers/produce.rs | 648 +++++++++++++++++++++-
gateways/kafka/tests/produce_real_bridge_tests.rs | 544 ++++++++++++++++++
gateways/kafka/tests/response_negative_tests.rs | 9 +-
12 files changed, 1359 insertions(+), 72 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index c18175ac4..c8428a4e4 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -39,7 +39,7 @@ max-threads = 1
filter = 'package(integration) and test(/connectors::elasticsearch::/)'
test-group = "elasticsearch"
-# Each test in this binary spawns its own real iggy-server subprocess (isolated data dir, port
+# Each test in these binaries spawns its own real iggy-server subprocess (isolated data dir, port
# from PortGuard's own flock, so no two ever share a port). #[serial] on them is a dead letter
# under nextest - nextest runs every test as its own process, and #[serial]'s mutex only excludes
# within one process (nextest does not support #[serial] macro, same gap the workaround above
@@ -55,8 +55,14 @@ test-group = "elasticsearch"
[test-groups.kafka_bridge]
max-threads = 4
+# Every gateway suite that spawns a server belongs here. A new one left out does not fail, it
+# just runs unbounded alongside these, which is the failure nobody notices until CI is slow.
[[profile.default.overrides]]
-filter = 'binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests) or binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)'
+filter = '''
+ binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)
+ + binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::produce_real_bridge_tests)
+'''
test-group = "kafka_bridge"
[profile.default]
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 66b509459..d65743919 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -2,13 +2,13 @@
Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests, validates scoped API keys and versions, and returns stub responses.
-> **Stub warning:** most APIs still don't persist or read real data. Produce and Fetch return
-> retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere instead of
-> trusting a fake success. CreateTopics does **not** create topics; valid requests return
-> `NOT_CONTROLLER` (41). Metadata still reports requested topics as unknown. ListOffsets is wired
-> to the Iggy bridge: with `IGGY_KAFKA_BRIDGE_ENABLED=true` it answers `EARLIEST`/`LATEST` from
-> real partition state; with the bridge off (the default) it stays a stub and answers
-> `NOT_LEADER_OR_FOLLOWER` (6). See [docs/SCOPE.md](docs/SCOPE.md).
+> **Stub warning:** When you set `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce writes to Iggy and
+> ListOffsets answers `EARLIEST`/`LATEST` from real partition state. No other API stores or reads
+> real data. Produce and ListOffsets without a bridge, and Fetch with or without one, answer
+> retriable `NOT_LEADER_OR_FOLLOWER` (6). Clients then keep their data and do not trust a fake
+> success. CreateTopics answers `NOT_CONTROLLER` (41) and creates nothing. Metadata reports every
+> topic as unknown, so a real client cannot reach Produce or ListOffsets yet. See
+> [docs/SCOPE.md](docs/SCOPE.md).
## Run
@@ -87,10 +87,41 @@ guards the hop from the gateway to Iggy rather than the hop from the producer to
`src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy
streams/topics, provisions them on demand, and looks up high watermarks (one or many partitions of
a topic per call) for `ListOffsets`.
-**Not wired into the live Produce/Fetch dispatch path yet** - that lands with
-[#3535](https://github.com/apache/iggy/issues/3535)/[#3536](https://github.com/apache/iggy/issues/3536).
-Exercised today by `bridge`'s own unit tests and `tests/bridge_iggy_integration_tests.rs` (spawns a
-real `iggy-server`).
+ListOffsets ([#3537](https://github.com/apache/iggy/issues/3537)) and Produce
+([#3535](https://github.com/apache/iggy/issues/3535), see below) call it. Fetch does not call it
+yet ([#3536](https://github.com/apache/iggy/issues/3536)).
+Tested by `bridge`'s own unit tests, `tests/bridge_iggy_integration_tests.rs`,
+`tests/list_offsets_real_bridge_tests.rs` and `tests/produce_real_bridge_tests.rs`. The last three
+start a real `iggy-server`.
+
+### Produce ([#3535](https://github.com/apache/iggy/issues/3535))
+
+Each partition entry in a Produce request becomes one Iggy `send_messages` call. Each entry
+answers for itself. A partition that fails carries its own error code, and the partitions that
+worked keep their offsets.
+
+| Field | What the gateway does |
+| ------- | ----------------------- |
+| Partition | Uses the index the client sent. Both systems number partitions from 0. |
+| Base offset | Takes it from the send confirmation. Sends `-1` when the server confirms the write but names no offset. |
+| `log_start_offset` | Always `-1`. The real value costs one extra call per request, and no producer reads it yet. |
+| `acks` | Writes the same way for `0`, `1` and `-1`. Durability is the Iggy topic's setting. Any other value answers `INVALID_REQUIRED_ACKS` (21). |
+| Topics | Creates none. A missing topic answers `UNKNOWN_TOPIC_OR_PARTITION` (3). Metadata creates topics. |
+| `timeout_ms` | Honors it, up to 20 seconds. Partitions past the deadline answer `REQUEST_TIMED_OUT` (7). |
+| Compression | Reads gzip, snappy, lz4 and zstd. |
+
+`acks=0` still writes the records. It answers nothing, because the wire protocol forbids a reply.
+
+Bad records get one of two codes. A record Iggy cannot hold answers `MESSAGE_TOO_LARGE` (10). A
+record this gateway cannot map answers `INVALID_RECORD` (87). A transactional or control batch
+answers `UNSUPPORTED_VERSION` (35), which is the code that stops a transactional producer. None of
+the three is retriable, because the same bytes never succeed on a second try.
+
+All batches in one request share one decompression budget, set to `max_frame_size`. A compressed
+request therefore cannot produce more bytes than the same client could have sent uncompressed. A
+request past the budget answers `MESSAGE_TOO_LARGE` (10).
+
+[docs/BRIDGE_MAPPING.md](docs/BRIDGE_MAPPING.md) describes what a record becomes once it is stored.
### Connection config
@@ -117,11 +148,11 @@ a detached background task specifically so that dropping the awaiting future - w
does on expiry - cannot abort it mid-flight. A timed-out call can leave that task holding the
shared client's connection lock for up to another 30s (the SDK's own reply deadline), queuing
every other bridge call behind it. `IggyBridge` holds one `IggyClient` with no pooling (see
-Concurrency ceiling below) and no semaphore bounding concurrent bridge calls - no longer a
-hypothetical now that ListOffsets (`#3537`) calls it from a real handler; more pressing once
-CreateTopics (`#3538`), Metadata (`#3534`), Produce (`#3535`) and Fetch (`#3536`) add their own
-concurrent callers. See `IggyBridge`'s own doc comment (its rustdoc is private, so this isn't a
-followable link outside the crate - read the source at `src/bridge/iggy_bridge.rs`).
+Concurrency ceiling below) and no semaphore bounding concurrent bridge calls. ListOffsets
+(`#3537`) and Produce (`#3535`) call it from real handlers, Produce on every write, so this is no
+longer a hypothetical. CreateTopics (`#3538`), Metadata (`#3534`) and Fetch (`#3536`) add more
+concurrent callers when they land. See `IggyBridge`'s own doc comment (its rustdoc is private, so
+this isn't a followable link outside the crate - read the source at `src/bridge/iggy_bridge.rs`).
### Topic mapping
@@ -180,15 +211,22 @@ One `IggyBridge` (one `IggyClient`) is meant to serve every Kafka connection thi
and the Iggy SDK's TCP transport is lockstep - one request in flight per client, its stream mutex
held across write, flush, and read. Every concurrent Kafka connection ends up serialized behind
whichever single Iggy request is in flight; the Kafka side's own connection limit
-(`IGGY_KAFKA_MAX_CONNECTIONS`) does nothing to relieve this. No connection pooling exists yet - it
-is a known gap to address before `#3535`/`#3536` put this on a hot path, not a design decision to
-rely on.
+(`IGGY_KAFKA_MAX_CONNECTIONS`) does nothing to relieve this. No connection pooling exists yet, and
+Produce puts this on a hot path as of `#3535` - it is a known gap, not a design decision to rely
+on.
+
+Message order does not depend on the pool. The connection loop reads one frame, runs the handler,
+then writes the answer. One Kafka connection therefore never has two requests in flight, and a
+producer's writes to a partition stay in the order it sent them. The single client costs
+throughput instead. A producer set up for several requests in flight gains nothing from them, and
+every Kafka connection waits behind the Iggy call that is running.
### Error mapping
`BridgeError::to_kafka_error_code()` maps Iggy failures to Kafka wire error codes:
-- Stream/topic/partition not found → `UNKNOWN_TOPIC_OR_PARTITION` (3)
+- Stream, topic or partition not found → `UNKNOWN_TOPIC_OR_PARTITION` (3). This includes the
+ generic `ResourceNotFound` that a partition request returns when the server cannot resolve it
- A rejected *permission* (`Unauthorized`) → `TOPIC_AUTHORIZATION_FAILED` (29) - a real,
fixable-by-the-Kafka-operator ACL problem
- A rejected *login* (the bridge's own `IGGY_KAFKA_IGGY_USERNAME`/`_PASSWORD` are wrong) →
diff --git a/gateways/kafka/docs/BRIDGE_MAPPING.md b/gateways/kafka/docs/BRIDGE_MAPPING.md
index 8791feffb..a4004edbe 100644
--- a/gateways/kafka/docs/BRIDGE_MAPPING.md
+++ b/gateways/kafka/docs/BRIDGE_MAPPING.md
@@ -282,9 +282,11 @@ states its output size up front, and `snap` reads that number without allocating
size is charged before the decoder runs, for Kafka's own block framing and for raw snappy alike.
`records::decode_batches` decompresses, and `records::DecompressionBudget` is the bound. The
-budget is a parameter, so setting it to `max_frame_size` for the whole request belongs to the
-Produce handler in [#3535](https://github.com/apache/iggy/issues/3535). Until that lands, nothing
-calls either one in a server path.
+budget is a parameter, and the Produce handler
+([#3535](https://github.com/apache/iggy/issues/3535)) sets it to `max_frame_size` once per
+request. That handler converts every partition entry before it writes any of them. The budget
+holds its remaining allowance in a `Cell`, so a reference to it held across an await makes every
+connection task `!Send`. Converting first keeps the budget out of the async half.
## Offsets
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index ee4bb749b..b9b7cf904 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -70,13 +70,19 @@ The gateway binds `:9093` by default; `kafka-message-gen`'s own default is `:909
broker's usual port). Every command below passes `--host 127.0.0.1:9093` explicitly - omitting it
connection-refuses before any assertion runs.
+Every case below describes a gateway started **without** a bridge (`IGGY_KAFKA_BRIDGE_ENABLED`
+unset). That is the only mode where every API answers with a stub. With a bridge, Produce writes
+to Iggy instead, and the empty record blob that the fixture tool sends answers `ec=87`
+(INVALID_RECORD) and not `ec=6`. To drive that path by hand you need real record batches, so
+`tests/produce_real_bridge_tests.rs` covers it against a real `iggy-server`.
+
| ID | Test | Steps | Expected result | Pass criteria |
| ---- | ------ | ------- | ----------------- | --------------- |
| A1 | Gateway starts | Run `iggy-gateway-kafka` | Binds to `:9093`, no panic | Log shows bind address |
| A2 | ApiVersions v1 | `cargo run -p kafka-message-gen -- send --host 127.0.0.1:9093 --api-key 18 --version 1` | Response received | `ec=0`, non-zero byte count |
| A3 | ApiVersions v3 (flexible) | Same with `--version 3` | Response received | `ec=0` |
| A4 | Metadata v0 | `send --host 127.0.0.1:9093 --api-key 3 --version 0` | Stub broker in response | Topic entries show `ec=3` (UNKNOWN_TOPIC_OR_PARTITION, stub) |
-| A5 | Produce v3 | `send --host 127.0.0.1:9093 --api-key 0 --version 3` | Decode + stub retriable error | `ec=6` (NOT_LEADER_OR_FOLLOWER) per partition |
+| A5 | Produce v3 | `send --host 127.0.0.1:9093 --api-key 0 --version 3` | Decode + stub retriable error, no bridge | `ec=6` (NOT_LEADER_OR_FOLLOWER) per partition |
| A6 | Fetch v4 | `send --host 127.0.0.1:9093 --api-key 1 --version 4` | Decode + stub response | Top-level `ec=0`; per-partition `ec=6` (NOT_LEADER_OR_FOLLOWER) |
| A7 | ListOffsets v1 | `send --host 127.0.0.1:9093 --api-key 2 --version 1` | Decode + stub offsets | Per-partition `ec=6` (NOT_LEADER_OR_FOLLOWER) - no top-level error field on this response |
| A8 | CreateTopics v2 | `send --host 127.0.0.1:9093 --api-key 19 --version 2` | Decode + stub non-creation ack | `ec=41` (NOT_CONTROLLER) per topic |
@@ -132,8 +138,8 @@ that one connection) is still serving other clients.
Run every `send` below with `--host 127.0.0.1:9093`. This category validates that the wire
encoding round-trips at the legacy/flexible boundary - not the stub error code, which is
-per-API and constant across both rows (see Category A: `ec=6` for Produce, `ec=41` for
-CreateTopics, and so on).
+per-API and constant across both rows (see Category A: `ec=6` for Produce with no bridge, `ec=41`
+for CreateTopics, and so on).
| ID | API key | Version | Encoding | Validation |
| ---- | --------- | --------- | ---------- | ------------ |
@@ -175,7 +181,7 @@ Requires `kcat` installed. Gateway does **not** implement SASL or full broker se
| ID | Test | Command | Expected (foundation) |
| ---- | ------ | --------- | --------------------- |
| G1 | Broker metadata | `kcat -b 127.0.0.1:9093 -L` | ApiVersions + Metadata handshake; broker appears in metadata |
-| G2 | Produce (likely fails later) | `echo "hello" \| kcat -b 127.0.0.1:9093 -t test -P` | May fail at coordinator/group stage — document actual error |
+| G2 | Produce (fails at metadata) | `echo "hello" \| kcat -b 127.0.0.1:9093 -t test -P` | Fails before it sends a Produce request. Metadata reports every topic as unknown, so the client finds no leader. [#3534](https://github.com/apache/iggy/issues/3534) unblocks this, not the Produce handler |
| G3 | Consumer (likely fails later) | `kcat -b 127.0.0.1:9093 -t test -C -o beginning` | May fail without consumer groups — document actual error |
Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke.
@@ -198,8 +204,11 @@ Record kcat version and exact error strings in your test log. G1 passing is the
| ------ | ------ | --------------- |
| 0 | NONE | Fetch top-level error field only (`ec=0` there does not mean per-partition success - see A6) |
| 6 | NOT_LEADER_OR_FOLLOWER | Produce/Fetch/ListOffsets stub, per partition (retriable; payload not persisted) |
+| 10 | MESSAGE_TOO_LARGE | Produce with a bridge: a record Iggy cannot hold, or a request that decompresses past its budget |
+| 21 | INVALID_REQUIRED_ACKS | Produce with a bridge: `acks` is not 0, 1 or -1 |
+| 87 | INVALID_RECORD | Produce with a bridge: a record batch this gateway cannot map, including a missing or empty one |
| 3 | UNKNOWN_TOPIC_OR_PARTITION | Metadata stub, per topic |
-| 35 | UNSUPPORTED_VERSION | **ApiVersions only** (KIP-511 exception). Every other API key's out-of-range version closes the connection instead - see Category B |
+| 35 | UNSUPPORTED_VERSION | **ApiVersions only** (KIP-511 exception), plus a Produce with a bridge that carries a transactional or control batch. Every other API key's out-of-range version closes the connection instead - see Category B |
| 37 | INVALID_PARTITIONS | CreateTopics: partition count `0` or `< -1` (or any non-positive on v2–v3) |
| 38 | INVALID_REPLICATION_FACTOR | CreateTopics: replication factor `0` or `< -1` (or any non-positive on v2–v3) |
| 41 | NOT_CONTROLLER | CreateTopics stub (topic not created) |
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index ea05eda70..9669abda9 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -4,7 +4,7 @@
Foundation layer only: a TCP listener on the Kafka wire port that decodes requests, validates scoped API keys and versions, validates request wire formats, and returns stub responses. **No Iggy backend integration.**
-**Stub semantics (important):** Produce discards the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). CreateTopics validates the request but answers with `NOT_CONTROLLER` (41) so clients do not believe topics were created. Do not treat `ec=0` stub success as durable storage — that arrives in the Iggy bridge phase.
+**Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). CreateTopics validates the request but answers `NOT_CONTROLLER` (41), so clients do not believe topics were created. Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores real data once you configure a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
| Deliverable | Status | Location |
| ------------- | -------- | ---------- |
@@ -44,7 +44,7 @@ it knows the server supports flexible encoding.
| --------- | ------ | ------------- | ------------- | ---------------- | ---------- |
| 18 | ApiVersions | 0 | 3 | 0, 1, 2, 3 | Advertise supported ranges; flexible encoding at v3+ |
| 3 | Metadata | 0 | 9 | 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 | Decode topic list count; stub broker host from `advertised_host` or the bound `local_addr` IP; flexible encoding at v9+ |
-| 0 | Produce | 3 | 9 | 3, 4, 5, 6, 7, 8, 9 | Decode request; stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
+| 0 | Produce | 3 | 9 | 3, 4, 5, 6, 7, 8, 9 | With a bridge: one `send_messages` per partition entry. Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | Decode request; stub response |
| 2 | ListOffsets | 1 | 6 | 1, 2, 3, 4, 5, 6 | Decode request; stub response |
| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default on v4+ |
@@ -89,7 +89,7 @@ Full reference for future phases: [`kafka_api_keys_reference.md`](kafka_api_keys
| ------- | ------- | ------------- |
| **1 — Wire framing** | In scope | `server.rs` — custom, zero-copy frame I/O; `header.rs` delegates version selection to `kafka_protocol::messages::ApiKey` |
| **2 — Request/response codecs** | Partial | Decode/encode via the `kafka_protocol` crate (broker feature only) for 6 hot-path keys; `bounds_guard.rs` pre-validates against unbounded allocation before handing a frame to the crate; stub responses only |
-| **3 — Iggy bridge** | Landed, not wired in | `bridge/` module (connection, topic mapping, provisioning, high watermark) landed; Produce/Fetch handler wiring itself is a follow-on ([#3535](https://github.com/apache/iggy/issues/3535)/[#3536](https://github.com/apache/iggy/issues/3536)) |
+| **3 — Iggy bridge** | Produce and ListOffsets wired | `bridge/` module (connection, topic mapping, provisioning, high watermark, `send_records`). Produce ([#3535](https://github.com/apache/iggy/issues/3535)) and ListOffsets ([#3537](https://github.com/apache/iggy/issues/3537)) call it. Fetch does not call it yet ([#3536](https://github.com/apache/iggy/issues/3536)) |
---
@@ -103,10 +103,11 @@ Items from the [hybrid architecture review](https://github.com/apache/iggy/discu
below it are still open for the issues that build on top of it.
- [x] Add `bridge/` module (`iggy_bridge`) - connection lifecycle, topic mapping, provisioning,
- high watermark, error mapping. See [README.md](../README.md#iggy-bridge-3533). Produce →
- `send_messages` / Fetch → `poll_messages` handler wiring itself is
- [#3535](https://github.com/apache/iggy/issues/3535)/[#3536](https://github.com/apache/iggy/issues/3536),
- not part of `bridge/`'s own scope.
+ high watermark, error mapping. See [README.md](../README.md#iggy-bridge-3533).
+- [x] Produce → `send_messages` ([#3535](https://github.com/apache/iggy/issues/3535)) - one call
+ per partition entry, base offset from the send confirmation, one error code per partition.
+ See [README.md](../README.md#produce-3535).
+- [ ] Fetch → `poll_messages` ([#3536](https://github.com/apache/iggy/issues/3536)).
- [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`.
- [x] Document partition mapping in [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md):
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index 9a5361cab..8d4db0f45 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -20,10 +20,10 @@ suite goes through `tests/common/fixtures.rs::load_fixture_body_or_skip`, which
regeneration hint when a fixture is missing, and panics instead when `KAFKA_FIXTURES_REQUIRED=1`
is set (CI sets this) so a broken generation step can't leave a suite green with zero assertions.
-### `iggy-server` binary (required for `bridge_iggy_integration_tests`)
+### `iggy-server` binary (required for `bridge_iggy_integration_tests`, `list_offsets_real_bridge_tests` and `produce_real_bridge_tests`)
-No fixtures needed, but `iggy-server` has to be built *first* - this suite spawns it directly and
-does not build it for you:
+No fixtures needed, but `iggy-server` has to be built *first* - these suites spawn it directly and
+do not build it for you:
```bash
cargo build --package server --bin iggy-server
@@ -64,9 +64,10 @@ file under `tests/` anymore.
| [`server_e2e_tests.rs`](../tests/server_e2e_tests.rs) | Full `KafkaGateway` TCP round-trips | Partial |
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases | No (needs the `iggy-server` binary - see Prerequisites) |
+| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) |
`tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `scope.rs`, `server.rs`,
-`tcp.rs`, `wire.rs`), compiled per test binary via `#[path]`, not a test binary itself. `codec.rs`
+`iggy_server.rs`, `tcp.rs`, `wire.rs`), compiled per test binary via `#[path]`, not a test binary itself. `codec.rs`
is test-only primitive encode/decode scaffolding for hand-building legacy/adversarial wire shapes
`kafka_protocol`'s spec-correct encoder cannot produce - it is not the gateway's production codec.
diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs
index 10f2035e3..d88ea1db4 100644
--- a/gateways/kafka/src/bridge/error.rs
+++ b/gateways/kafka/src/bridge/error.rs
@@ -127,6 +127,12 @@ impl BridgeError {
/// `ClientState::Connected`, the ordinary window between a reconnect's TCP handshake completing
/// and its auto-sign-in landing, not for a rejected login.
///
+/// `ResourceNotFound` joins them despite naming no resource. `dispatch_partition_request`
+/// (`core/server/src/dispatch/partition.rs`) flattens every unresolved stream, topic and
+/// partition into it, so a `send_messages` to a missing topic or partition arrives as this and
+/// never as the typed variants, which only the bridge's own `get_topic` paths build locally.
+/// Every resource this bridge addresses is one of those three.
+///
/// `TransientNotAccepted` (Iggy replica-side "retry, on any replica") is retriable the same way.
/// `TransientNotCommitted` is not folded into that set: its outcome is genuinely unknown rather
/// than known-safe-to-retry, so it maps to `REQUEST_TIMED_OUT` instead of
@@ -152,7 +158,8 @@ const fn iggy_error_to_kafka_code(err: &IggyError) -> i16 {
| IggyError::StreamNameNotFound(_)
| IggyError::TopicIdNotFound(_, _)
| IggyError::TopicNameNotFound(_, _)
- | IggyError::PartitionNotFound(_, _, _) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ | IggyError::PartitionNotFound(_, _, _)
+ | IggyError::ResourceNotFound(_) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
IggyError::Unauthorized => ERROR_TOPIC_AUTHORIZATION_FAILED,
IggyError::Disconnected
| IggyError::EmptyResponse
@@ -194,6 +201,14 @@ mod tests {
assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_TOPIC_OR_PARTITION);
}
+ #[test]
+ fn resource_not_found_maps_to_unknown_topic_or_partition() {
+ // The only code a `send_messages` to a missing topic or partition ever comes back as:
+ // the server flattens both to this one rather than to the typed variants above.
+ let err = BridgeError::Iggy(IggyError::ResourceNotFound(String::new()));
+ assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
#[test]
fn unauthenticated_maps_to_not_leader_or_follower_for_retry() {
// Not a credential rejection: fail_if_not_authenticated (binary_auth.rs) returns this for
diff --git a/gateways/kafka/src/bridge/iggy_bridge/produce.rs b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
index 453732212..97150f51c 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/produce.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
@@ -17,5 +17,63 @@
//! Produce-side bridge calls.
//!
-//! Empty until the Produce handler lands. Separate from `fetch.rs` so the two never edit one
-//! file.
+//! Separate from `fetch.rs` so the two never edit one file.
+
+use iggy::prelude::{Identifier, IggyMessage, MessageClient, Partitioning};
+
+use super::{IggyBridge, with_request_timeout};
+use crate::bridge::error::BridgeError;
+use crate::bridge::topic_map::validate_kafka_topic_name;
+
+impl IggyBridge {
+ /// Appends `messages` to one partition of the Iggy topic `kafka_topic` maps to, and reports
+ /// where the batch landed.
+ ///
+ /// Takes the Kafka-side name so a topic with a mapping override reaches the same Iggy
+ /// resource the other bridge calls do. `partition` passes through unconverted, since both
+ /// systems number partitions from 0 and the client already chose one. An index the topic
+ /// does not have is rejected by the server, so this spends no round trip checking first.
+ ///
+ /// Creates nothing. A producer asks `Metadata` to create a missing topic before it produces
+ /// to it, so auto-creation belongs to that handler.
+ ///
+ /// `Ok(None)` is a committed send the server named no offset for, which happens for a
+ /// request it classified as a duplicate and for a batch whose journal entry is gone. The
+ /// caller answers `-1`. Retrying would duplicate a write that already committed.
+ ///
+ /// The first confirmation naming `partition` wins: the server reports one per send, and a
+ /// reply carrying several would carry them in send order.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::InvalidKafkaTopicName`] if the name fails Kafka's own rules.
+ /// [`BridgeError::Timeout`] if the send outruns `REQUEST_TIMEOUT`, which leaves the outcome
+ /// unknown rather than known-failed. [`BridgeError::Iggy`] if the stream, topic or partition
+ /// is missing, the message is too large, or the connection failed.
+ pub async fn send_records(
+ &self,
+ kafka_topic: &str,
+ partition: u32,
+ messages: &mut [IggyMessage],
+ ) -> Result<Option<u64>, BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ let topic_id = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ let partitioning = Partitioning::partition_id(partition);
+
+ let response = with_request_timeout(self.client.send_messages(
+ &stream_id,
+ &topic_id,
+ &partitioning,
+ messages,
+ ))
+ .await?;
+
+ Ok(response
+ .confirmations
+ .iter()
+ .find(|confirmation| confirmation.partition_id == partition)
+ .map(|confirmation| confirmation.base_offset))
+ }
+}
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index 3c29a9512..cd8bca7df 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -51,12 +51,18 @@ pub const ERROR_NOT_LEADER_OR_FOLLOWER: i16 = 6;
/// real regardless of which retriable code is sent; it closes only once `#3535` has an idempotent
/// produce path, not by picking a different error code here.
pub const ERROR_REQUEST_TIMED_OUT: i16 = 7;
+/// Produce: a record over Iggy's `MAX_PAYLOAD_SIZE`, or a request decompressing past its budget.
+/// Terminal, correctly: the same bytes never fit on a second try.
+pub const ERROR_MESSAGE_TOO_LARGE: i16 = 10;
/// `bridge`'s mapping for a Kafka-side topic name that fails Kafka's own naming rules.
///
/// Empty, whitespace-padded, over 249 bytes, or outside `[A-Za-z0-9._-]`, checked before any Iggy
/// call is made - a real Kafka client library validates topic names client-side and would never
/// send one of these, but a raw/non-conformant client could.
pub const ERROR_INVALID_TOPIC_EXCEPTION: i16 = 17;
+/// Produce: `acks` is not 0, 1 or -1. A conformant client never sends one, since `acks` comes
+/// from validated configuration rather than from application input.
+pub const ERROR_INVALID_REQUIRED_ACKS: i16 = 21;
/// Closest fit for an Iggy permission/credential rejection in `bridge`'s error mapping.
///
/// There is no bridge-side SASL exchange yet (`#3549`), so `SASL_AUTHENTICATION_FAILED` would
@@ -82,6 +88,12 @@ pub const ERROR_INVALID_REQUEST: i16 = 42;
/// Non-retriable, so a Java client resolves immediately instead of retrying
/// [`ERROR_UNKNOWN_SERVER_ERROR`] until its own `default.api.timeout.ms`.
pub const ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT: i16 = 43;
+/// Produce: a record or batch this gateway cannot map.
+///
+/// Not `CORRUPT_MESSAGE` (2), whose text fits but which `kafka-protocol`'s table marks
+/// retriable, so a client would resend a batch that can never decode. A client older than Kafka
+/// 2.4 reads 87 as a generic server error, which is still terminal and still better than a loop.
+pub const ERROR_INVALID_RECORD: i16 = 87;
/// Result of handling one Kafka request body.
#[derive(Debug)]
diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs
index 944b72503..5b4abc43f 100644
--- a/gateways/kafka/src/protocol/handlers/produce.rs
+++ b/gateways/kafka/src/protocol/handlers/produce.rs
@@ -17,19 +17,27 @@
//! Produce (API key 0).
+use std::time::Duration;
+
use bytes::Bytes;
+use iggy::prelude::{IggyError, IggyMessage};
+use kafka_protocol::messages::produce_request::PartitionProduceData;
use kafka_protocol::messages::produce_response::{PartitionProduceResponse, TopicProduceResponse};
-use kafka_protocol::messages::{ProduceRequest, ProduceResponse};
+use kafka_protocol::messages::{ProduceRequest, ProduceResponse, TopicName};
+use crate::bridge::IggyBridge;
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_PRODUCE, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNSUPPORTED_VERSION,
- GatewayState, HandleOutcome, supported_max_version,
+ API_KEY_PRODUCE, ApiVersionRange, ERROR_INVALID_RECORD, ERROR_INVALID_REQUIRED_ACKS,
+ ERROR_MESSAGE_TOO_LARGE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_REQUEST_TIMED_OUT,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
+ supported_max_version,
};
use crate::protocol::bounds_guard::validate_produce_shape;
use crate::protocol::handlers::{
decode_guarded, encode_message, respond_or_close, unsupported_version_response,
};
+use crate::records::{DecompressionBudget, RecordCodecError, decode_batches, to_iggy};
pub const RANGE: ApiVersionRange = ApiVersionRange {
api_key: API_KEY_PRODUCE,
@@ -37,6 +45,40 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
max_version: 9,
};
+/// The three `acks` values the protocol defines. Anything else is a malformed request.
+const ACKS_NONE: i16 = 0;
+const ACKS_LEADER: i16 = 1;
+const ACKS_ALL: i16 = -1;
+
+/// Kafka's "no offset here", used for `base_offset` on any failure, for a committed send the
+/// server named no offset for, and for `log_start_offset` always. The real log start costs a
+/// `get_topic` round trip per request and no producer reads it yet.
+const UNKNOWN_OFFSET: i64 = -1;
+
+/// Ceiling on how long one request may spend writing, whatever the client asked for.
+///
+/// One request is many Iggy calls, each with the bridge's own 15s timeout, so without this a
+/// request naming many partitions has no bound. Deliberately a clock and not a partition-count
+/// cap: a cap refuses a producer writing to a wide topic, and a client set on hogging the shared
+/// bridge just sends more requests instead.
+const MAX_REQUEST_DEADLINE: Duration = Duration::from_secs(20);
+
+const RESPONSE_BASE_BYTES: usize = 512;
+const RESPONSE_BYTES_PER_PARTITION: usize = 64;
+
+/// A converted partition entry: where to write it, or the code it already answers with.
+type PartitionPlan = std::result::Result<(u32, Vec<IggyMessage>), i16>;
+
+struct PlannedPartition {
+ index: i32,
+ plan: PartitionPlan,
+}
+
+struct PlannedTopic {
+ name: TopicName,
+ partitions: Vec<PlannedPartition>,
+}
+
/// Produce is the only request the wire protocol allows to go unanswered
/// (`acks=0`), so it gets its own path that may return [`HandleOutcome::NoResponse`].
///
@@ -45,10 +87,6 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
/// spec-compliant client can legitimately send Produce v0-2 with `acks=0`. Rejecting those
/// versions before reading `acks` would send an error response the client never expects,
/// desyncing the next correlation id it reads.
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until a handler awaits the bridge"
-)]
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
// Above the encoder max there is no response parseable at the client's version, so close
// rather than decode (same policy the other APIs apply). Fail-closed on a missing row
@@ -70,22 +108,16 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
_ => None,
};
return match acks {
- Some(0) | None => HandleOutcome::NoResponse,
+ Some(ACKS_NONE) | None => HandleOutcome::NoResponse,
Some(_) => unsupported_version_response(API_KEY_PRODUCE, api_version, |v| {
encode_error_response(v, ERROR_UNSUPPORTED_VERSION)
}),
};
}
- match decode_guarded::<ProduceRequest>(api_version, body, |v, b| {
+ let request = match decode_guarded::<ProduceRequest>(api_version, body, |v, b| {
validate_produce_shape(v, b, state.max_frame_size)
}) {
- // acks=0 is fire-and-forget: the client isn't reading a response, so
- // sending one desyncs the next correlation id it expects.
- Ok(req) if req.acks == 0 => HandleOutcome::NoResponse,
- // `api_version` is always in `[3, supported_max]` here: the `< 3` case returned above,
- // and the `> max` case returned at the top of this function - so it is always within
- // `SUPPORTED_RANGES`' Produce row and an `is_supported_version` re-check can never fail.
- Ok(req) => respond_or_close(encode_response(api_version, &req), "Produce"),
+ Ok(request) => request,
Err(error) => {
// `kafka_protocol` decodes the whole request in one shot; a failure anywhere gives
// no partial-field access, so `acks` is unknowable here (unlike the pre-migration
@@ -98,12 +130,268 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
// a client looping malformed bodies on one connection (never disconnected - this
// arm returns NoResponse) has no rate limit here.
tracing::debug!(%error, "failed to decode Produce request (no response)");
- HandleOutcome::NoResponse
+ return HandleOutcome::NoResponse;
+ }
+ };
+
+ let Some(bridge) = state.bridge.as_deref() else {
+ return stub_outcome(api_version, &request);
+ };
+
+ // `acks=0` is legal, so a request failing this check is one the client waits on.
+ if !matches!(request.acks, ACKS_NONE | ACKS_LEADER | ACKS_ALL) {
+ tracing::debug!(
+ acks = request.acks,
+ "Produce request names an unknown acks value"
+ );
+ return respond_or_close(
+ encode_uniform_response(api_version, &request, ERROR_INVALID_REQUIRED_ACKS),
+ "Produce",
+ );
+ }
+
+ // Before planning, not after: the client waits through decompression too.
+ let deadline = tokio::time::Instant::now() + request_timeout(request.timeout_ms);
+
+ // Convert everything, then write it. The budget holds its allowance in a `Cell`, so a
+ // reference to it live across an await would make every connection task `!Send`. The scope
+ // drops it before the first one.
+ //
+ // One budget for the whole request, not one per batch: a request carries up to
+ // `MAX_REQUEST_ELEMENTS` partition entries, so a per-batch cap would still admit that
+ // multiple of it.
+ let plan = {
+ let budget = DecompressionBudget::new(state.max_frame_size);
+ plan_request(&budget, &request)
+ };
+ let responses = write_plan(bridge, plan, deadline).await;
+
+ // The write already ran. Answering `NoResponse` before it would drop every record an
+ // `acks=0` producer sends, silently and at full speed.
+ if request.acks == ACKS_NONE {
+ return HandleOutcome::NoResponse;
+ }
+ respond_or_close(encode_written_response(api_version, responses), "Produce")
+}
+
+/// No bridge: answer retriable so clients keep their data instead of trusting a discarded write.
+fn stub_outcome(api_version: i16, request: &ProduceRequest) -> HandleOutcome {
+ // acks=0 is fire-and-forget: the client isn't reading a response, so
+ // sending one desyncs the next correlation id it expects.
+ if request.acks == ACKS_NONE {
+ return HandleOutcome::NoResponse;
+ }
+ respond_or_close(encode_response(api_version, request), "Produce")
+}
+
+/// Converts every partition entry. Synchronous, so the budget never crosses an await.
+fn plan_request(budget: &DecompressionBudget, request: &ProduceRequest) -> Vec<PlannedTopic> {
+ request
+ .topic_data
+ .iter()
+ .map(|topic| PlannedTopic {
+ name: topic.name.clone(),
+ partitions: topic
+ .partition_data
+ .iter()
+ .map(|partition| PlannedPartition {
+ index: partition.index,
+ plan: plan_partition(budget, topic.name.as_str(), partition),
+ })
+ .collect(),
+ })
+ .collect()
+}
+
+/// One entry's batches in, its messages out. A batch that cannot be mapped is refused whole,
+/// never half-stored.
+fn plan_partition(
+ budget: &DecompressionBudget,
+ kafka_topic: &str,
+ partition: &PartitionProduceData,
+) -> PartitionPlan {
+ // The records are not what is wrong with a negative index, so not `INVALID_RECORD`.
+ let Ok(partition_id) = u32::try_from(partition.index) else {
+ return Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ };
+ // Nothing to append. Answering success would name an offset for a write that never happened.
+ let Some(blob) = partition.records.as_ref() else {
+ return Err(ERROR_INVALID_RECORD);
+ };
+
+ // The clone is a refcount, and `decode_batches` drains what it is given.
+ let mut blob = blob.clone();
+ let records = decode_batches(&mut blob, budget).map_err(|error| {
+ tracing::debug!(%error, kafka_topic, partition_id, "refused a Produce record batch");
+ record_error_code(&error)
+ })?;
+ // An empty blob decodes to zero batches rather than to an error, so this rules on it.
+ if records.is_empty() {
+ return Err(ERROR_INVALID_RECORD);
+ }
+
+ let mut messages = Vec::with_capacity(records.len());
+ for record in &records {
+ messages.push(to_iggy(record).map_err(|error| {
+ tracing::debug!(%error, kafka_topic, partition_id, "refused a Kafka record");
+ record_error_code(&error)
+ })?);
+ }
+ Ok((partition_id, messages))
+}
+
+/// The shorter of the client's `timeout_ms` and this gateway's ceiling.
+///
+/// Honoring the client's budget is what stops its retry from duplicating a write still running
+/// here. A non-positive value means no preference, not "fail without trying".
+fn request_timeout(timeout_ms: i32) -> Duration {
+ u64::try_from(timeout_ms)
+ .ok()
+ .filter(|ms| *ms > 0)
+ .map_or(MAX_REQUEST_DEADLINE, Duration::from_millis)
+ .min(MAX_REQUEST_DEADLINE)
+}
+
+/// Appends every planned partition, in request order.
+///
+/// Sequential: the one `IggyClient` behind the bridge is lockstep, so concurrent sends would
+/// just queue on its stream mutex (see the README's "Concurrency ceiling").
+async fn write_plan(
+ bridge: &IggyBridge,
+ plan: Vec<PlannedTopic>,
+ deadline: tokio::time::Instant,
+) -> Vec<TopicProduceResponse> {
+ let mut topics = Vec::with_capacity(plan.len());
+ for PlannedTopic { name, partitions } in plan {
+ let mut responses = Vec::with_capacity(partitions.len());
+ for PlannedPartition { index, plan } in partitions {
+ responses.push(write_partition(bridge, name.as_str(), index, plan, deadline).await);
}
+ topics.push(
+ TopicProduceResponse::default()
+ .with_name(name)
+ .with_partition_responses(responses),
+ );
}
+ topics
}
-/// Well-formed Produce response with a single placeholder topic/partition.
+async fn write_partition(
+ bridge: &IggyBridge,
+ kafka_topic: &str,
+ index: i32,
+ plan: PartitionPlan,
+ deadline: tokio::time::Instant,
+) -> PartitionProduceResponse {
+ let outcome = match plan {
+ Ok((partition_id, mut messages)) => {
+ send_by(bridge, kafka_topic, partition_id, &mut messages, deadline).await
+ }
+ Err(error_code) => Err(error_code),
+ };
+ match outcome {
+ // An offset past `i64::MAX` has no response field to fit in, so it reads as unknown
+ // rather than as a saturated number a client would take for a real position.
+ Ok(base_offset) => partition_response(index, ERROR_NONE).with_base_offset(
+ base_offset.map_or(UNKNOWN_OFFSET, |offset| {
+ i64::try_from(offset).unwrap_or(UNKNOWN_OFFSET)
+ }),
+ ),
+ Err(error_code) => partition_response(index, error_code),
+ }
+}
+
+/// Appends one partition's messages, or answers 7 once the deadline has passed.
+///
+/// 7 and not a connection-shaped code in both timeout arms: the SDK writes on a detached task
+/// this timeout cannot abort, so a send we stop waiting on may still land. 7 is the code for an
+/// outcome nobody knows, and retrying on it may duplicate the write.
+async fn send_by(
+ bridge: &IggyBridge,
+ kafka_topic: &str,
+ partition_id: u32,
+ messages: &mut [IggyMessage],
+ deadline: tokio::time::Instant,
+) -> std::result::Result<Option<u64>, i16> {
+ let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+ if remaining.is_zero() {
+ tracing::warn!(
+ kafka_topic,
+ partition_id,
+ "Produce deadline passed before this partition was written"
+ );
+ return Err(ERROR_REQUEST_TIMED_OUT);
+ }
+
+ let send = bridge.send_records(kafka_topic, partition_id, messages);
+ match tokio::time::timeout(remaining, send).await {
+ Ok(Ok(base_offset)) => Ok(base_offset),
+ Ok(Err(error)) => {
+ // warn!, not debug!: an operator can act on this, and the client only sees the code.
+ tracing::warn!(%error, kafka_topic, partition_id, "Iggy refused a Produce batch");
+ Err(error.to_kafka_error_code())
+ }
+ Err(_elapsed) => {
+ tracing::warn!(
+ kafka_topic,
+ partition_id,
+ "Produce deadline passed while this partition was being written"
+ );
+ Err(ERROR_REQUEST_TIMED_OUT)
+ }
+ }
+}
+
+/// The Kafka code one record or batch failure answers with.
+///
+/// Two-way split: too big is 10, malformed is 87. Not `CORRUPT_MESSAGE` (2) for the malformed
+/// half, whose text fits but which `kafka_protocol` marks retriable, so a client would resend
+/// bytes that can never decode. Both codes here are terminal.
+///
+/// No wildcard arm, so a variant added later fails the match instead of taking a default.
+const fn record_error_code(error: &RecordCodecError) -> i16 {
+ match error {
+ RecordCodecError::Iggy(error) => message_error_code(error),
+ RecordCodecError::BudgetExceeded { .. } | RecordCodecError::EnvelopeTooLarge { .. } => {
+ ERROR_MESSAGE_TOO_LARGE
+ }
+ // Both kinds mean transactions: only a transaction produces a control batch. 35 is the
+ // one code a transactional producer treats as fatal, so it stops instead of aborting and
+ // retrying forever.
+ RecordCodecError::UnsupportedBatch(_) => ERROR_UNSUPPORTED_VERSION,
+ // Reachable from Produce.
+ RecordCodecError::TimestampOutOfRange(_)
+ | RecordCodecError::Batch(_)
+ | RecordCodecError::RecordCountTooLarge { .. }
+ | RecordCodecError::HeaderCountTooLarge { .. }
+ | RecordCodecError::RecordTruncated
+ | RecordCodecError::RecordFieldLength(_)
+ // Raised by `from_iggy` only, so Fetch is what actually answers for these.
+ | RecordCodecError::UserHeadersUnreadable(_)
+ | RecordCodecError::MappingVersion(_)
+ | RecordCodecError::ValueMarker(_)
+ | RecordCodecError::HeaderNameCollision(_)
+ | RecordCodecError::TimestampMarker(_)
+ | RecordCodecError::EnvelopeTruncated { .. }
+ | RecordCodecError::EnvelopeTrailingBytes(_)
+ | RecordCodecError::EnvelopeVersion(_)
+ | RecordCodecError::EnvelopeHeaderName => ERROR_INVALID_RECORD,
+ }
+}
+
+/// The code for a message `IggyMessage::new` refused.
+///
+/// It has three rejections: an oversized payload, oversized headers, and an empty payload. The
+/// first two are size failures. The third is unreachable, since the codec writes a placeholder
+/// byte for a null or empty value.
+const fn message_error_code(error: &IggyError) -> i16 {
+ match error {
+ IggyError::TooBigMessagePayload | IggyError::TooBigUserHeaders => ERROR_MESSAGE_TOO_LARGE,
+ _ => ERROR_INVALID_RECORD,
+ }
+}
+
+/// Well-formed response with a single placeholder topic and partition.
///
/// # Errors
///
@@ -113,16 +401,21 @@ pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
TopicProduceResponse::default()
.with_partition_responses(vec![partition_response(0, error_code)]),
]);
- encode_message(&resp, version, 512)
+ encode_message(&resp, version, RESPONSE_BASE_BYTES)
}
-/// Stub: discard the payload and return a retriable error so clients keep data locally until
-/// the Iggy bridge lands (do not advertise silent success).
+/// Stub: discard the payload and answer retriable, rather than advertise a silent success.
///
/// # Errors
///
/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
pub fn encode_response(version: i16, req: &ProduceRequest) -> Result<Bytes> {
+ encode_uniform_response(version, req, ERROR_NOT_LEADER_OR_FOLLOWER)
+}
+
+/// One code for every partition named, for failures that belong to the request and not to a
+/// single partition.
+fn encode_uniform_response(version: i16, req: &ProduceRequest, error_code: i16) -> Result<Bytes> {
let responses = req
.topic_data
.iter()
@@ -133,18 +426,325 @@ pub fn encode_response(version: i16, req: &ProduceRequest) -> Result<Bytes> {
topic
.partition_data
.iter()
- .map(|p| partition_response(p.index, ERROR_NOT_LEADER_OR_FOLLOWER))
+ .map(|p| partition_response(p.index, error_code))
.collect(),
)
})
.collect();
let resp = ProduceResponse::default().with_responses(responses);
- encode_message(&resp, version, 512)
+ encode_message(&resp, version, RESPONSE_BASE_BYTES)
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+fn encode_written_response(version: i16, responses: Vec<TopicProduceResponse>) -> Result<Bytes> {
+ let partitions: usize = responses
+ .iter()
+ .map(|topic| topic.partition_responses.len())
+ .sum();
+ let capacity =
+ RESPONSE_BASE_BYTES.saturating_add(partitions.saturating_mul(RESPONSE_BYTES_PER_PARTITION));
+ let resp = ProduceResponse::default().with_responses(responses);
+ encode_message(&resp, version, capacity)
}
fn partition_response(index: i32, error_code: i16) -> PartitionProduceResponse {
PartitionProduceResponse::default()
.with_index(index)
.with_error_code(error_code)
- .with_log_start_offset(0)
+ .with_base_offset(UNKNOWN_OFFSET)
+ .with_log_start_offset(UNKNOWN_OFFSET)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use bytes::BytesMut;
+ use kafka_protocol::indexmap::IndexMap;
+ use kafka_protocol::messages::produce_request::TopicProduceData;
+ use kafka_protocol::protocol::StrBytes;
+ use kafka_protocol::records::{
+ Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID, NO_SEQUENCE,
+ Record, RecordBatchEncoder, RecordEncodeOptions, TimestampType,
+ };
+
+ use crate::records::encode_batch;
+
+ const CREATE_TIME: i64 = 1_700_000_000_123;
+ const TOPIC: &str = "orders";
+ /// Well above anything these batches decompress to.
+ const TEST_BUDGET: usize = 1024 * 1024;
+
+ fn record(offset: i64, value: &[u8]) -> Record {
+ Record {
+ transactional: false,
+ control: false,
+ delete_horizon: false,
+ partition_leader_epoch: NO_PARTITION_LEADER_EPOCH,
+ producer_id: NO_PRODUCER_ID,
+ producer_epoch: NO_PRODUCER_EPOCH,
+ timestamp_type: TimestampType::Creation,
+ offset,
+ sequence: NO_SEQUENCE,
+ timestamp: CREATE_TIME,
+ key: None,
+ value: Some(Bytes::copy_from_slice(value)),
+ headers: IndexMap::new(),
+ }
+ }
+
+ fn gzip_batch(records: &[Record]) -> Bytes {
+ let mut buf = BytesMut::new();
+ let options = RecordEncodeOptions {
+ version: 2,
+ compression: Compression::Gzip,
+ };
+ RecordBatchEncoder::encode(&mut buf, records.iter(), &options).expect("encode gzip batch");
+ buf.freeze()
+ }
+
+ fn entry(index: i32, records: Option<Bytes>) -> PartitionProduceData {
+ PartitionProduceData::default()
+ .with_index(index)
+ .with_records(records)
+ }
+
+ fn plan(entry: &PartitionProduceData) -> PartitionPlan {
+ plan_partition(&DecompressionBudget::new(TEST_BUDGET), TOPIC, entry)
+ }
+
+ #[test]
+ fn given_a_record_batch_when_planned_should_convert_every_record_in_order() {
+ let batch = encode_batch(&mut [record(0, b"first"), record(1, b"second")]).unwrap();
+ let (partition_id, messages) = plan(&entry(2, Some(batch))).expect("batch must convert");
+
+ assert_eq!(
+ partition_id, 2,
+ "the Kafka index passes through unconverted"
+ );
+ assert_eq!(messages.len(), 2, "one record becomes one message");
+ assert_eq!(messages[0].payload.as_ref(), b"first");
+ assert_eq!(messages[1].payload.as_ref(), b"second");
+ }
+
+ #[test]
+ fn given_a_negative_partition_index_when_planned_should_answer_unknown_topic_or_partition() {
+ let batch = encode_batch(&mut [record(0, b"v")]).unwrap();
+ assert_eq!(
+ plan(&entry(-1, Some(batch))).unwrap_err(),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ "the records are not what is wrong with the request"
+ );
+ }
+
+ #[test]
+ fn given_no_records_field_when_planned_should_answer_invalid_record() {
+ assert_eq!(plan(&entry(0, None)).unwrap_err(), ERROR_INVALID_RECORD);
+ }
+
+ #[test]
+ fn given_an_empty_records_blob_when_planned_should_answer_invalid_record() {
+ // An empty blob reads as zero batches, not as a malformed one, so nothing below rejects it.
+ assert_eq!(
+ plan(&entry(0, Some(Bytes::new()))).unwrap_err(),
+ ERROR_INVALID_RECORD
+ );
+ }
+
+ #[test]
+ fn given_a_transactional_batch_when_planned_should_answer_unsupported_version() {
+ let mut records = [record(0, b"v")];
+ records[0].transactional = true;
+ let batch = encode_batch(&mut records).unwrap();
+
+ assert_eq!(
+ plan(&entry(0, Some(batch))).unwrap_err(),
+ ERROR_UNSUPPORTED_VERSION,
+ "35 is the only code a transactional producer treats as fatal"
+ );
+ }
+
+ #[test]
+ fn given_a_control_batch_when_planned_should_answer_unsupported_version() {
+ let mut records = [record(0, b"v")];
+ records[0].control = true;
+ let batch = encode_batch(&mut records).unwrap();
+
+ assert_eq!(
+ plan(&entry(0, Some(batch))).unwrap_err(),
+ ERROR_UNSUPPORTED_VERSION,
+ "a stored message carries no control flag, so this one cannot be admitted"
+ );
+ }
+
+ #[test]
+ fn given_a_truncated_batch_when_planned_should_answer_invalid_record() {
+ let batch = encode_batch(&mut [record(0, b"value")]).unwrap();
+ let truncated = batch.slice(..batch.len() - 3);
+
+ assert_eq!(
+ plan(&entry(0, Some(truncated))).unwrap_err(),
+ ERROR_INVALID_RECORD,
+ "not CORRUPT_MESSAGE (2), which is retriable and would loop"
+ );
+ }
+
+ #[test]
+ fn given_a_batch_that_decompresses_past_the_budget_when_planned_should_answer_too_large() {
+ let batch = gzip_batch(&[record(0, &[b'a'; 4096])]);
+ let budget = DecompressionBudget::new(16);
+
+ assert_eq!(
+ plan_partition(&budget, TOPIC, &entry(0, Some(batch))).unwrap_err(),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ }
+
+ #[test]
+ fn given_one_budget_when_two_partitions_are_planned_should_share_its_allowance() {
+ // Per request, not per batch: many entries must not each get the whole budget.
+ let budget = DecompressionBudget::new(6144);
+ let first = entry(0, Some(gzip_batch(&[record(0, &[b'a'; 4096])])));
+ let second = entry(1, Some(gzip_batch(&[record(0, &[b'b'; 4096])])));
+
+ assert!(plan_partition(&budget, TOPIC, &first).is_ok());
+ assert_eq!(
+ plan_partition(&budget, TOPIC, &second).unwrap_err(),
+ ERROR_MESSAGE_TOO_LARGE,
+ "the second batch must be charged against what the first one left"
+ );
+ }
+
+ #[test]
+ fn given_a_request_when_planned_should_keep_its_topic_and_partition_order() {
+ let batch = encode_batch(&mut [record(0, b"v")]).unwrap();
+ let request = ProduceRequest::default().with_topic_data(vec![
+ TopicProduceData::default()
+ .with_name(TopicName(StrBytes::from_static_str(TOPIC)))
+ .with_partition_data(vec![entry(1, Some(batch)), entry(0, None)]),
+ ]);
+
+ let planned = plan_request(&DecompressionBudget::new(TEST_BUDGET), &request);
+ assert_eq!(planned.len(), 1);
+ assert_eq!(planned[0].name.as_str(), TOPIC);
+ let indices: Vec<i32> = planned[0].partitions.iter().map(|p| p.index).collect();
+ assert_eq!(
+ indices,
+ vec![1, 0],
+ "the response follows the request order"
+ );
+ assert!(planned[0].partitions[0].plan.is_ok());
+ assert_eq!(
+ planned[0].partitions[1].plan.as_ref().unwrap_err(),
+ &ERROR_INVALID_RECORD,
+ "one partition failing must not cost the other its own answer"
+ );
+ }
+
+ #[test]
+ fn given_a_client_timeout_when_bounded_should_take_whichever_is_shorter() {
+ assert_eq!(request_timeout(5_000), Duration::from_secs(5));
+ assert_eq!(
+ request_timeout(30_000),
+ MAX_REQUEST_DEADLINE,
+ "a client asking for longer than this gateway allows gets the gateway's ceiling"
+ );
+ assert_eq!(request_timeout(i32::MAX), MAX_REQUEST_DEADLINE);
+ }
+
+ #[test]
+ fn given_no_client_timeout_when_bounded_should_still_attempt_the_write() {
+ // A value naming no budget means no preference, not "fail now".
+ assert_eq!(request_timeout(0), MAX_REQUEST_DEADLINE);
+ assert_eq!(request_timeout(-1), MAX_REQUEST_DEADLINE);
+ assert_eq!(request_timeout(i32::MIN), MAX_REQUEST_DEADLINE);
+ }
+
+ #[test]
+ fn given_a_partition_response_when_built_should_name_no_offset_and_no_append_time() {
+ let response = partition_response(3, ERROR_INVALID_RECORD);
+
+ assert_eq!(response.index, 3);
+ assert_eq!(response.error_code, ERROR_INVALID_RECORD);
+ assert_eq!(response.base_offset, UNKNOWN_OFFSET);
+ assert_eq!(response.log_start_offset, UNKNOWN_OFFSET);
+ assert_eq!(
+ response.log_append_time_ms, UNKNOWN_OFFSET,
+ "CreateTime, which is the only timestamp type the mapping stores"
+ );
+ }
+
+ #[test]
+ fn given_a_request_level_failure_when_encoded_should_answer_every_partition_it_named() {
+ let request = ProduceRequest::default().with_topic_data(vec![
+ TopicProduceData::default()
+ .with_name(TopicName(StrBytes::from_static_str(TOPIC)))
+ .with_partition_data(vec![entry(0, None), entry(1, None)]),
+ ]);
+
+ let body =
+ encode_uniform_response(3, &request, ERROR_INVALID_REQUIRED_ACKS).expect("encode");
+ assert!(!body.is_empty());
+ }
+
+ #[test]
+ fn given_a_size_failure_when_mapped_should_answer_message_too_large() {
+ assert_eq!(
+ record_error_code(&RecordCodecError::BudgetExceeded {
+ produced: 2,
+ remaining: 1
+ }),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ assert_eq!(
+ record_error_code(&RecordCodecError::EnvelopeTooLarge { size: 1 }),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ assert_eq!(
+ record_error_code(&RecordCodecError::Iggy(IggyError::TooBigMessagePayload)),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ assert_eq!(
+ record_error_code(&RecordCodecError::Iggy(IggyError::TooBigUserHeaders)),
+ ERROR_MESSAGE_TOO_LARGE
+ );
+ }
+
+ #[test]
+ fn given_a_malformed_record_when_mapped_should_answer_a_terminal_code() {
+ for error in [
+ RecordCodecError::Batch("bad".to_string()),
+ RecordCodecError::RecordTruncated,
+ RecordCodecError::RecordFieldLength(-1),
+ RecordCodecError::TimestampOutOfRange(i64::MIN),
+ RecordCodecError::RecordCountTooLarge {
+ count: i32::MAX,
+ limit: 1,
+ },
+ RecordCodecError::HeaderCountTooLarge {
+ count: i32::MAX,
+ limit: 1,
+ },
+ RecordCodecError::Iggy(IggyError::InvalidMessagePayloadLength),
+ ] {
+ assert_eq!(
+ record_error_code(&error),
+ ERROR_INVALID_RECORD,
+ "{error} must not answer a retriable code"
+ );
+ }
+ }
+
+ #[test]
+ fn given_a_refused_batch_kind_when_mapped_should_answer_unsupported_version() {
+ assert_eq!(
+ record_error_code(&RecordCodecError::UnsupportedBatch("transactional")),
+ ERROR_UNSUPPORTED_VERSION
+ );
+ assert_eq!(
+ record_error_code(&RecordCodecError::UnsupportedBatch("control")),
+ ERROR_UNSUPPORTED_VERSION
+ );
+ }
}
diff --git a/gateways/kafka/tests/produce_real_bridge_tests.rs b/gateways/kafka/tests/produce_real_bridge_tests.rs
new file mode 100644
index 000000000..45500d438
--- /dev/null
+++ b/gateways/kafka/tests/produce_real_bridge_tests.rs
@@ -0,0 +1,544 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Produce (API key 0) driven through the whole handler against a real `iggy-server`.
+//!
+//! Records go in as Kafka wire bytes and come back out through the Iggy SDK, so a mapping broken
+//! the same way in both directions cannot pass.
+
+use std::sync::Arc;
+
+use bytes::{Bytes, BytesMut};
+use iggy::prelude::{
+ Consumer, HeaderKey, Identifier, IggyMessage, MessageClient, PollingStrategy, TopicClient,
+};
+use kafka_protocol::indexmap::IndexMap;
+use kafka_protocol::protocol::StrBytes;
+use kafka_protocol::records::{
+ Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID, NO_SEQUENCE, Record,
+ RecordBatchEncoder, RecordEncodeOptions, TimestampType,
+};
+use serial_test::serial;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_PRODUCE, BrokerAdvertise, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, handle_request_bounded,
+};
+use iggy_gateway_kafka::records::{KEY_HEADER, MAPPING_VERSION, VERSION_HEADER, from_iggy};
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+
+use codec::{Decoder, Encoder};
+use iggy_server::{TestServer, raw_client};
+
+/// The stream every unmapped Kafka topic resolves to, per `TopicMapping`'s default rule.
+const STREAM: &str = "kafka";
+const TOPIC: &str = "orders";
+const CREATE_TIME: i64 = 1_700_000_000_123;
+const MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+
+/// Spelled out rather than imported: a test reading its expected code from the same constant the
+/// handler answers with cannot catch that constant changing.
+const ERROR_INVALID_REQUIRED_ACKS: i16 = 21;
+const ERROR_UNSUPPORTED_VERSION: i16 = 35;
+
+/// A live gateway with the topic already provisioned, since Produce creates nothing.
+async fn gateway_with_topic(server: &TestServer, partitions: u32) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ bridge
+ .ensure_stream_and_topic(TOPIC, partitions)
+ .await
+ .expect("the topic must exist before producing to it");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ MAX_FRAME_SIZE,
+ )
+}
+
+fn record(offset: i64, key: Option<&[u8]>, value: &[u8], headers: &[(&str, &[u8])]) -> Record {
+ Record {
+ transactional: false,
+ control: false,
+ delete_horizon: false,
+ partition_leader_epoch: NO_PARTITION_LEADER_EPOCH,
+ producer_id: NO_PRODUCER_ID,
+ producer_epoch: NO_PRODUCER_EPOCH,
+ timestamp_type: TimestampType::Creation,
+ offset,
+ sequence: NO_SEQUENCE,
+ timestamp: CREATE_TIME,
+ key: key.map(Bytes::copy_from_slice),
+ value: Some(Bytes::copy_from_slice(value)),
+ headers: headers
+ .iter()
+ .map(|(name, value)| {
+ (
+ StrBytes::from_string((*name).to_string()),
+ Some(Bytes::copy_from_slice(value)),
+ )
+ })
+ .collect::<IndexMap<_, _>>(),
+ }
+}
+
+fn batch(records: &[Record], compression: Compression) -> Bytes {
+ let mut buf = BytesMut::new();
+ let options = RecordEncodeOptions {
+ version: 2,
+ compression,
+ };
+ RecordBatchEncoder::encode(&mut buf, records.iter(), &options).expect("encode record batch");
+ buf.freeze()
+}
+
+/// One partition entry of a request: the index and the records blob it carries.
+type Entry = (i32, Bytes);
+
+/// Builds a request body by hand, at either wire encoding.
+///
+/// Not through `kafka_protocol`: the crate is taken with the `broker` feature alone, which
+/// encodes responses and decodes requests, neither of the halves a client needs. Hand-built
+/// bytes also make these tests exercise the real decoder instead of round-tripping one schema.
+fn encode_request(version: i16, acks: i16, topic: &str, entries: &[Entry]) -> Bytes {
+ let flexible = version >= 9;
+ let mut enc = Encoder::with_capacity(1024);
+
+ if flexible {
+ enc.write_compact_nullable_string(None); // transactional_id
+ } else {
+ enc.write_nullable_string(None).expect("null fits");
+ }
+ enc.write_i16(acks);
+ enc.write_i32(5_000); // timeout_ms
+
+ if flexible {
+ enc.write_varint(2); // one topic, compact array is N+1
+ enc.write_compact_nullable_string(Some(topic));
+ enc.write_varint(entries.len() as u64 + 1);
+ } else {
+ enc.write_i32(1);
+ enc.write_nullable_string(Some(topic)).expect("name fits");
+ enc.write_i32(i32::try_from(entries.len()).expect("few entries"));
+ }
+
+ for (index, records) in entries {
+ enc.write_i32(*index);
+ if flexible {
+ enc.write_compact_nullable_bytes(Some(records));
+ enc.write_empty_tagged_fields();
+ } else {
+ enc.write_nullable_bytes(Some(records))
+ .expect("records fit");
+ }
+ }
+ if flexible {
+ enc.write_empty_tagged_fields(); // topic
+ enc.write_empty_tagged_fields(); // request
+ }
+ enc.freeze()
+}
+
+/// Every partition entry of a response, flattened, in the order the response lists them.
+fn partition_results(version: i16, body: Bytes) -> Vec<(i32, i16, i64)> {
+ let flexible = version >= 9;
+ let mut d = Decoder::new(body);
+ let topics = read_count(&mut d, flexible);
+
+ let mut results = Vec::new();
+ for _ in 0..topics {
+ read_name(&mut d, flexible);
+ let partitions = read_count(&mut d, flexible);
+ for _ in 0..partitions {
+ let index = d.read_i32().expect("partition index");
+ let error_code = d.read_i16().expect("error code");
+ let base_offset = d.read_i64().expect("base offset");
+ let append_time = d.read_i64().expect("log append time");
+ assert_eq!(append_time, -1, "CreateTime topics report no append time");
+ if version >= 5 {
+ d.read_i64().expect("log start offset");
+ }
+ if version >= 8 {
+ assert_eq!(read_count(&mut d, flexible), 0, "no per-record errors");
+ read_name(&mut d, flexible); // error_message
+ }
+ if flexible {
+ d.read_tagged_fields().expect("partition tagged fields");
+ }
+ results.push((index, error_code, base_offset));
+ }
+ if flexible {
+ d.read_tagged_fields().expect("topic tagged fields");
+ }
+ }
+
+ assert_eq!(d.read_i32().expect("throttle time"), 0);
+ if flexible {
+ d.read_tagged_fields().expect("response tagged fields");
+ }
+ assert_eq!(d.remaining(), 0, "the response must decode exactly");
+ results
+}
+
+fn read_count(d: &mut Decoder, flexible: bool) -> usize {
+ if flexible {
+ let raw = d.read_varint().expect("compact array length");
+ usize::try_from(raw.saturating_sub(1)).expect("array length fits")
+ } else {
+ usize::try_from(d.read_i32().expect("array length")).expect("array length fits")
+ }
+}
+
+fn read_name(d: &mut Decoder, flexible: bool) -> Option<String> {
+ if flexible {
+ d.read_compact_nullable_string().expect("compact string")
+ } else {
+ d.read_nullable_string().expect("string")
+ }
+}
+
+/// Sends one request through the handler and returns the partition entries it answered with.
+async fn produce(
+ state: &GatewayState,
+ version: i16,
+ acks: i16,
+ topic: &str,
+ entries: &[Entry],
+) -> Vec<(i32, i16, i64)> {
+ let body = encode_request(version, acks, topic, entries);
+ let response = handle_request_bounded(state, API_KEY_PRODUCE, version, body)
+ .await
+ .expect_response("a Produce request with acks != 0 must be answered");
+ partition_results(version, response)
+}
+
+async fn stored(server: &TestServer, partition: u32, count: u32) -> Vec<IggyMessage> {
+ raw_client(server)
+ .await
+ .poll_messages(
+ &Identifier::named(STREAM).expect("stream name"),
+ &Identifier::named(TOPIC).expect("topic name"),
+ Some(partition),
+ &Consumer::new(Identifier::named("reader").expect("consumer name")),
+ &PollingStrategy::offset(0),
+ count,
+ false,
+ )
+ .await
+ .expect("poll the partition back")
+ .messages
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_produce_request_when_handled_should_store_every_record_it_carried() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let records = [
+ record(0, Some(b"k1"), b"first", &[("trace", b"abc")]),
+ record(1, None, b"second", &[]),
+ ];
+ let entries = [(0, batch(&records, Compression::None))];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![(0, ERROR_NONE, 0)],
+ "a two-record batch lands at offset 0 and reports it"
+ );
+
+ let messages = stored(&server, 0, 10).await;
+ assert_eq!(
+ messages.len(),
+ 2,
+ "one Kafka record became one Iggy message"
+ );
+ assert_eq!(messages[0].payload.as_ref(), b"first");
+ assert_eq!(messages[1].payload.as_ref(), b"second");
+
+ // Once on the raw stored shape, since a round trip through the codec proves nothing about
+ // the shape itself.
+ let headers = messages[0]
+ .user_headers_map()
+ .expect("headers must parse")
+ .expect("a gateway-written message always carries headers");
+ assert_eq!(
+ headers
+ .get(&HeaderKey::try_from(VERSION_HEADER).unwrap())
+ .map(|value| value.as_bytes().to_vec()),
+ Some(vec![MAPPING_VERSION]),
+ "the provenance header marks this message as gateway-written"
+ );
+ assert_eq!(
+ headers
+ .get(&HeaderKey::try_from(KEY_HEADER).unwrap())
+ .map(|value| value.as_bytes().to_vec()),
+ Some(b"k1".to_vec()),
+ "the Kafka key is stored beside the value, not inside it"
+ );
+
+ let back = from_iggy(&messages[0], 0).expect("stored message must decode as a record");
+ assert_eq!(back.value.as_deref(), Some(&b"first"[..]));
+ assert_eq!(back.timestamp, CREATE_TIME, "CreateTime survives the store");
+ assert_eq!(
+ back.headers
+ .get(&StrBytes::from_static_str("trace"))
+ .and_then(Option::as_deref),
+ Some(&b"abc"[..])
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_flexible_produce_request_when_handled_should_store_its_records() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let records = [record(0, Some(b"k"), b"flexible", &[])];
+ let entries = [(0, batch(&records, Compression::None))];
+
+ assert_eq!(
+ produce(&state, 9, -1, TOPIC, &entries).await,
+ vec![(0, ERROR_NONE, 0)],
+ "the compact and tagged-field encoding reaches the same write path"
+ );
+ assert_eq!(stored(&server, 0, 10).await.len(), 1);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_gzip_batch_when_handled_should_store_its_records() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let records = [record(0, None, b"compressed", &[])];
+ let entries = [(0, batch(&records, Compression::Gzip))];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![(0, ERROR_NONE, 0)]
+ );
+ assert_eq!(
+ stored(&server, 0, 10).await[0].payload.as_ref(),
+ b"compressed"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_two_batches_in_one_blob_when_handled_should_land_under_one_base_offset() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ // One `records` field may hold several batches back to back.
+ let mut blob = BytesMut::new();
+ blob.extend_from_slice(&batch(&[record(0, None, b"a", &[])], Compression::None));
+ blob.extend_from_slice(&batch(&[record(0, None, b"b", &[])], Compression::None));
+ let entries = [(0, blob.freeze())];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![(0, ERROR_NONE, 0)],
+ "both batches append as one Iggy send under one base offset"
+ );
+ let messages = stored(&server, 0, 10).await;
+ assert_eq!(messages.len(), 2);
+ assert_eq!(messages[1].payload.as_ref(), b"b");
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_second_request_when_handled_should_report_where_it_landed() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let first = [record(0, None, b"a", &[]), record(1, None, b"b", &[])];
+ let second = [record(0, None, b"c", &[])];
+
+ assert_eq!(
+ produce(
+ &state,
+ 3,
+ 1,
+ TOPIC,
+ &[(0, batch(&first, Compression::None))]
+ )
+ .await,
+ vec![(0, ERROR_NONE, 0)]
+ );
+ assert_eq!(
+ produce(
+ &state,
+ 3,
+ 1,
+ TOPIC,
+ &[(0, batch(&second, Compression::None))]
+ )
+ .await,
+ vec![(0, ERROR_NONE, 2)],
+ "the base offset comes from the send confirmation, not from a count this handler kept"
+ );
+ assert_eq!(stored(&server, 0, 10).await.len(), 3);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_partition_the_topic_lacks_when_handled_should_keep_the_other_answers() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 2).await;
+
+ let blob = batch(&[record(0, None, b"v", &[])], Compression::None);
+ let entries = [(0, blob.clone()), (9, blob.clone()), (1, blob)];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![
+ (0, ERROR_NONE, 0),
+ (9, ERROR_UNKNOWN_TOPIC_OR_PARTITION, -1),
+ (1, ERROR_NONE, 0),
+ ],
+ "one partition failing costs the others nothing, and a failure names no offset"
+ );
+ assert_eq!(stored(&server, 0, 10).await.len(), 1);
+ assert_eq!(stored(&server, 1, 10).await.len(), 1);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_an_unknown_topic_when_handled_should_answer_unknown_topic_or_partition() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let entries = [(0, batch(&[record(0, None, b"v", &[])], Compression::None))];
+
+ assert_eq!(
+ produce(&state, 3, 1, "never-created", &entries).await,
+ vec![(0, ERROR_UNKNOWN_TOPIC_OR_PARTITION, -1)],
+ "Produce provisions nothing: auto-creation belongs to Metadata"
+ );
+
+ let topics = raw_client(&server)
+ .await
+ .get_topics(&Identifier::named(STREAM).expect("stream name"))
+ .await
+ .expect("list the topics the stream holds");
+ assert_eq!(
+ topics.len(),
+ 1,
+ "the refused topic must not have been created on the way"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_acks_zero_when_handled_should_write_the_records_and_answer_nothing() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let entries = [(
+ 0,
+ batch(
+ &[record(0, None, b"fire-and-forget", &[])],
+ Compression::None,
+ ),
+ )];
+ let body = encode_request(3, 0, TOPIC, &entries);
+
+ let outcome = handle_request_bounded(&state, API_KEY_PRODUCE, 3, body).await;
+ assert!(
+ outcome.is_no_response(),
+ "the wire protocol forbids answering an acks=0 request"
+ );
+
+ let messages = stored(&server, 0, 10).await;
+ assert_eq!(
+ messages.len(),
+ 1,
+ "answering nothing is not the same as storing nothing"
+ );
+ assert_eq!(messages[0].payload.as_ref(), b"fire-and-forget");
+}
+
+#[tokio::test]
+#[serial]
+async fn given_an_unknown_acks_value_when_handled_should_answer_invalid_required_acks() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let entries = [(0, batch(&[record(0, None, b"v", &[])], Compression::None))];
+
+ assert_eq!(
+ produce(&state, 3, 2, TOPIC, &entries).await,
+ vec![(0, ERROR_INVALID_REQUIRED_ACKS, -1)]
+ );
+ assert!(
+ stored(&server, 0, 10).await.is_empty(),
+ "a request refused before the write must leave the partition empty"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_transactional_batch_when_handled_should_answer_unsupported_version() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ let mut records = [record(0, None, b"v", &[])];
+ records[0].transactional = true;
+ let entries = [(0, batch(&records, Compression::None))];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![(0, ERROR_UNSUPPORTED_VERSION, -1)],
+ "transactions are out of scope, and 35 is the code that stops a transactional producer"
+ );
+ assert!(
+ stored(&server, 0, 10).await.is_empty(),
+ "a refused batch must not be half-stored"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_no_bridge_when_handled_should_still_answer_the_retriable_stub() {
+ // Every other suite drives this path, so a regression in it surfaces far from Produce.
+ // Pinned here, next to the path that replaced it.
+ let state = GatewayState::stub(BrokerAdvertise::default(), MAX_FRAME_SIZE);
+ let entries = [(0, batch(&[record(0, None, b"v", &[])], Compression::None))];
+
+ assert_eq!(
+ produce(&state, 3, 1, TOPIC, &entries).await,
+ vec![(0, ERROR_NOT_LEADER_OR_FOLLOWER, -1)],
+ "without a bridge nothing is stored, so the answer must stay retriable"
+ );
+}
diff --git a/gateways/kafka/tests/response_negative_tests.rs b/gateways/kafka/tests/response_negative_tests.rs
index db05f69d4..14750559a 100644
--- a/gateways/kafka/tests/response_negative_tests.rs
+++ b/gateways/kafka/tests/response_negative_tests.rs
@@ -233,11 +233,12 @@ fn produce_error_response_v9_uses_flexible_record_errors_shape() {
Some(String::new())
);
assert_eq!(d.read_varint().unwrap(), 2); // one partition
- assert_eq!(d.read_i32().unwrap(), 0);
+ assert_eq!(d.read_i32().unwrap(), 0); // partition index
assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST);
- assert_eq!(d.read_i64().unwrap(), 0);
- assert_eq!(d.read_i64().unwrap(), -1);
- assert_eq!(d.read_i64().unwrap(), 0);
+ // An error appended nothing, so there is no offset to name and none is invented.
+ assert_eq!(d.read_i64().unwrap(), -1); // base_offset
+ assert_eq!(d.read_i64().unwrap(), -1); // log_append_time_ms, CreateTime
+ assert_eq!(d.read_i64().unwrap(), -1); // log_start_offset, not looked up
assert_eq!(d.read_varint().unwrap(), 1); // empty record_errors array
assert_eq!(d.read_compact_nullable_string().unwrap(), None);
}