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/29 05:21:58 UTC
(iggy) branch kafka-fetch updated (b2ef6f392 -> c5e897759)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch kafka-fetch
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard b2ef6f392 docs(gateways): stop calling Kafka Fetch a stub
discard 3761fa948 fix(gateways): keep Kafka Fetch from skipping records or starving slots
discard e06c1fd25 fix(gateways): make Kafka Fetch wait out Iggy loads and honor min_bytes
discard d6a1b3631 fix(gateways): stop Kafka Fetch skipping records after a failed read
discard b809b9173 feat(gateways): answer Kafka Fetch with records from Iggy
new c5e897759 feat(gateways): answer Kafka Fetch with records from Iggy
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b2ef6f392)
\
N -- N -- N refs/heads/kafka-fetch (c5e897759)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.config/nextest.toml | 2 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 211 ++++++++++++++++++++++++-
3 files changed, 206 insertions(+), 9 deletions(-)
(iggy) 01/01: feat(gateways): answer Kafka Fetch with records from Iggy
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-fetch
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c5e897759b5964f2ca2dcbd677b8f53972407562
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Sep 29 10:48:53 2026 +0530
feat(gateways): answer Kafka Fetch with records from Iggy
Fetch polls Iggy per partition and honors max_wait_ms and min_bytes.
Each read slot polls on its own client. E2e tests use kcat and Java.
---
.config/nextest.toml | 3 +-
gateways/kafka/Cargo.toml | 2 +
gateways/kafka/README.md | 120 +-
gateways/kafka/docs/ACL_MAPPING.md | 6 +-
gateways/kafka/docs/AUTHENTICATION.md | 10 +-
gateways/kafka/docs/BRIDGE_MAPPING.md | 10 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 4 +-
gateways/kafka/docs/MANUAL_TESTING.md | 2 +-
gateways/kafka/docs/SCOPE.md | 22 +-
gateways/kafka/docs/TEST_SUITE.md | 6 +-
gateways/kafka/src/bridge/error.rs | 4 +
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 368 +++-
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 75 +-
gateways/kafka/src/bridge/iggy_bridge/offsets.rs | 138 +-
gateways/kafka/src/bridge/iggy_bridge/produce.rs | 33 +-
gateways/kafka/src/bridge/mod.rs | 9 +-
gateways/kafka/src/protocol/api.rs | 29 +-
gateways/kafka/src/protocol/handlers/fetch.rs | 2020 +++++++++++++++++++-
.../kafka/src/protocol/handlers/list_offsets.rs | 102 +-
gateways/kafka/src/protocol/handlers/metadata.rs | 14 +-
gateways/kafka/src/protocol/handlers/mod.rs | 77 +-
gateways/kafka/src/protocol/handlers/produce.rs | 57 +-
gateways/kafka/src/protocol/mod.rs | 1 +
gateways/kafka/src/protocol/probe_board.rs | 368 ++++
gateways/kafka/src/records.rs | 103 +-
gateways/kafka/tests/fetch_real_bridge_tests.rs | 727 +++++++
gateways/kafka/tests/kafka_client_e2e_tests.rs | 211 +-
27 files changed, 4120 insertions(+), 401 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 4cb32eb3c..d9d69fbca 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -60,13 +60,14 @@ max-threads = 4
[[profile.default.overrides]]
filter = '''
binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)
+ + binary_id(iggy-gateway-kafka::fetch_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::produce_real_bridge_tests)
'''
test-group = "kafka_bridge"
# The real-client suite spawns its own server too, so it needs serializing for the same reason, but
-# in its own group rather than queued behind the bridge tests: sharing one group would put six
+# in its own group rather than queued behind the bridge tests: sharing one group would put eight
# container-driven tests behind twenty-one unrelated ones, serially, for no isolation gained. Both
# groups cap their servers' shard pools themselves, so running the two alongside each other is
# bounded.
diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml
index 11b3f1610..962656513 100644
--- a/gateways/kafka/Cargo.toml
+++ b/gateways/kafka/Cargo.toml
@@ -70,6 +70,8 @@ zstd = { workspace = true }
[dev-dependencies]
# Repairs the CRC of a batch header a test rewrote, which the decoder checks first.
crc32c = { workspace = true }
+# The client half, so a test encodes requests and decodes responses the way a consumer does.
+kafka-protocol = { workspace = true, features = ["client"] }
serial_test = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = [
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 4b4bcb708..8eb16bb38 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -1,19 +1,14 @@
# Kafka gateway (`iggy-gateway-kafka`)
-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. With a bridge, Produce writes to Iggy and ListOffsets reads offsets from it. Everything else is a stub.
-
-> **Stub warning:** Fetch does not read real data yet: with or without a bridge it returns
-> retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere instead of
-> trusting a fake success. With `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce writes to Iggy,
-> CreateTopics creates a real Iggy stream/topic, Metadata reports real topics and partition counts
-> (a topic not requested by name and not found is silently absent from a null-topics "list all"
-> response, and `UNKNOWN_TOPIC_OR_PARTITION` when named explicitly), and ListOffsets answers
-> `EARLIEST`/`LATEST` from real partition state. With the bridge off (the default) these stay stubs: Produce and
-> ListOffsets answer `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics answers `NOT_CONTROLLER` (41), and
-> Metadata reports every requested topic unknown. **CreateTopics runs as the bridge's own Iggy
-> user**: with the bridge on and `IGGY_KAFKA_SASL_ENABLED` off (the default), any client that can
-> reach this port can create topics (up to 1000 partitions each). See
-> [docs/SCOPE.md](docs/SCOPE.md).
+Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy. InitProducerId and consumer group coordination work with or without one.
+
+> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata and
+> CreateTopics use Iggy. With the bridge off (the default), those five are stubs: Produce, Fetch and
+> ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics answers `NOT_CONTROLLER`
+> (41), and Metadata reports every requested topic unknown. **CreateTopics runs as the bridge's own
+> Iggy user**: with the bridge on and `IGGY_KAFKA_SASL_ENABLED` off (the default), any client that
+> can reach this port can create topics (up to 1000 partitions each). Every client reads and writes
+> as the bridge's Iggy user. See [docs/SCOPE.md](docs/SCOPE.md).
>
> InitProducerId does real work too, with or without the bridge: it allocates a producer id, so a stock idempotent producer starts instead of failing at startup.
>
@@ -21,8 +16,9 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> `LeaveGroup` and `SyncGroup` are real, with real membership, rebalances, graceful leave and
> session expiry ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). With the bridge off,
> Metadata reports every topic unknown, so a consumer joins a group and is assigned 0 partitions.
-> With it on, partitions are assigned, but offset commit/fetch is not implemented and Fetch is
-> still a stub, so nothing can be consumed yet.
+> Offset commit and fetch are not implemented yet, so a consumer must use `assign()` with explicit
+> start offsets and `enable.auto.commit=false`. The Java client needs no `group.id`. librdkafka
+> refuses `assign()` without one, so set any value there.
## Run
@@ -156,10 +152,9 @@ Four things to know before switching it on:
`IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` bounds the checks the gateway runs at once, and a peer
whose login was rejected is refused for a delay that doubles per rejection, from 0.5s up to 30s.
- **Authentication and an ACL view, not enforcement.** The gateway verifies the credentials and
- can describe what Iggy grants the principal, but nothing gates an operation yet. Iggy's
- permissions decide that once Produce and Fetch are wired to it
- ([#3535](https://github.com/apache/iggy/issues/3535),
- [#3536](https://github.com/apache/iggy/issues/3536)).
+ can describe what Iggy grants the principal, but no operation checks those grants. Produce and
+ Fetch act as the bridge's Iggy user for every client, so any client that signs in reads and
+ writes what that user can.
### ACLs
@@ -181,14 +176,13 @@ Full reasoning, including what was rejected and why, is in
## Iggy bridge ([#3533](https://github.com/apache/iggy/issues/3533))
`src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy
-streams/topics, provisions them on demand, and looks up high watermarks (one or many partitions of
-a topic per call) for `ListOffsets`.
-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`.
+streams/topics, provisions them on demand, looks up high watermarks (one or many partitions of
+a topic per call) for `ListOffsets`, and probes and polls partitions for Fetch.
+Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
+([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)) and CreateTopics
+([#3538](https://github.com/apache/iggy/issues/3538)) call it.
+Tested by `bridge`'s own unit tests, `tests/bridge_iggy_integration_tests.rs` and the
+`tests/*_real_bridge_tests.rs` suites. All but the unit tests start a real `iggy-server`.
### Produce ([#3535](https://github.com/apache/iggy/issues/3535))
@@ -230,6 +224,58 @@ decoded. 4 requests decode at once.
[docs/BRIDGE_MAPPING.md](docs/BRIDGE_MAPPING.md) describes what a record becomes once it is stored.
+### Fetch ([#3536](https://github.com/apache/iggy/issues/3536))
+
+Per request: one probe per topic, shared with other Fetches, then polls of each partition with
+records, in request order. One uncompressed batch per partition.
+
+| Field | Gateway |
+| ----- | ------- |
+| Partition, offset | As sent. Both systems count from 0. |
+| Iggy replica | The one on the node the bridge connects to. |
+| Offset below the oldest kept one | Reads from the oldest kept one. If none is kept, no records until the next write, because a cluster replica that lost records after a crash reads the same. Kafka answers `1` here, so `auto.offset.reset=none` gets no error. |
+| Purge | Iggy restarts the partition at offset 0, and Fetch cannot see it. Seek consumers to 0 after a purge. Otherwise writes past a consumer's old offset make it skip the records below. |
+| `high_watermark` | One past the last committed offset. |
+| `last_stable_offset` | Same as `high_watermark`. No transactions. |
+| `log_start_offset` | Always `-1`. |
+| Sessions | None opened. `session_id` is always `0`, so every request is a full one. |
+| `max_wait_ms` | Honored, max 18 s. |
+| `min_bytes` | Honored, as Kafka does: the request waits until the records read reach it. New bytes are estimated from Iggy's average message size. |
+| `max_bytes` | Honored, max `IGGY_KAFKA_MAX_FRAME_SIZE`. |
+| `partition_max_bytes` | Honored. |
+| First partition with records | At least one record, even past every byte limit (KIP-74). |
+| Records per partition | At most 1000 per request, polled up to 16 at a time. |
+| `isolation_level`, replica id, epochs, rack, forgotten topics | Ignored. |
+
+| Code | When | Java consumer |
+| ---- | ---- | ------------- |
+| `OFFSET_OUT_OF_RANGE` (1) | Offset below 0, or past `high_watermark`. A partition with no offset past 0 answers 6 instead for its first 30 s, because Iggy reports that while it loads or fences one. | Resets its offset (`auto.offset.reset`). |
+| `UNKNOWN_TOPIC_OR_PARTITION` (3) | Topic or partition not in Iggy, or a name Kafka refuses | Refreshes metadata, retries. |
+| `NOT_LEADER_OR_FOLLOWER` (6) | Iggy unreachable, too slow, not signed in, loading the partition, or returning no records where it reports some. Alone, it goes out after `max_wait_ms`. | Refreshes metadata, retries. |
+| `TOPIC_AUTHORIZATION_FAILED` (29) | The bridge's Iggy user lacks permission. Fetch reads as that user, not as the client. | Fails. |
+| `UNKNOWN_SERVER_ERROR` (-1) | Anything else, such as a stored message the gateway cannot map. Alone, it goes out after `max_wait_ms`. The partition is read again at most once per `max_wait_ms`. | Retries the same offset. |
+| `FETCH_SESSION_ID_NOT_FOUND` (70) | Top level. The request continues a session. | Sends a full request. |
+
+The Java consumer throws on any other partition code, so Fetch sends none.
+
+Limits:
+
+- 20 s per request, wait included, so a waiting Fetch fits the 25 s shutdown drain. Partitions
+ not read by then answer `0` with no records.
+- 4 requests read and encode at once. A waiting request, or a response on a slow socket, holds no
+ slot. So response memory grows with connections: `max_bytes` each, or one larger record.
+- A message the gateway cannot map stops the consumer at its offset. Records before it are served.
+ The log shows it once at `warn`.
+- Each slot polls on its own Iggy client, which connects at the slot's first poll. A poll ends at
+ the request deadline, so a partition that Iggy refuses holds up only its own slot.
+- Iggy polls by count, not bytes. One poll can load 16 of the largest messages: about 1 GB, and
+ about 4 GB over the 4 slots. It also costs polls: 1000 small records take 63. A byte cap on Iggy
+ polls would lift both.
+- A request loads at most 100 new topic probes per round, the oldest first. Other topics use an
+ older probe. A waiting request probes 100 topics per round, in turn.
+- A Produce through the gateway wakes a waiting Fetch of that topic at once. A native Iggy write
+ wakes it at the next probe, within about 200 ms.
+
### Connection config
| Variable | Default | Description |
@@ -246,10 +292,10 @@ Iggy SDK client's own default of unlimited retries, one dial per second, forever
attempt - retries included - is capped at `REQUEST_TIMEOUT` (15s) wall-clock, so `IggyBridge::connect`
fails in bounded time whether the address refuses the connection or silently drops it, instead of
blocking the calling task indefinitely. Every other bridge call (`ensure_stream_and_topic`,
-`high_watermark(s)`, `close`) carries the same `REQUEST_TIMEOUT` for the same reason: the SDK
-reconnects internally, mid-call, on a transport error, through the same undead-lined dial path - a
-bridge call made well after the initial connect can still hit this if Iggy becomes unreachable
-later.
+`high_watermark(s)`, `probe`, `close`) carries the same `REQUEST_TIMEOUT` for the same
+reason: the SDK reconnects internally, mid-call, on a transport error, through the same
+undead-lined dial path - a bridge call made well after the initial connect can still hit this if
+Iggy becomes unreachable later.
`send_records` uses the Produce deadline instead:
@@ -310,12 +356,16 @@ cut before that threshold is reached - worth knowing rather than discovering lat
### Concurrency ceiling
-- One `IggyClient` serves every Kafka connection, one Iggy request at a time.
-- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A client pool is a TODO in
+- One `IggyClient` serves every Kafka connection, one Iggy request at a time. Fetch polls use 4
+ more, one per read slot.
+- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is a TODO in
[docs/SCOPE.md](docs/SCOPE.md).
- Order: set `max.in.flight.requests.per.connection=1`, or `retries=0`. Otherwise a retried batch
can land after later ones.
+Fetches share topic probes: one `get_topic` per topic at a time, for all consumers together, on
+that same client. A caught-up consumer of a busy topic can start one per Fetch.
+
### Error mapping
`BridgeError::to_kafka_error_code()` maps Iggy failures to Kafka wire error codes:
@@ -345,6 +395,8 @@ cut before that threshold is reached - worth knowing rather than discovering lat
through `ensure_topic`'s `partition_count` argument once it exceeds the server's cap
- Anything else → `UNKNOWN_SERVER_ERROR` (-1)
+Fetch folds these into the codes a consumer handles: 7 → 6, 17 → 3, and any other code → -1.
+
### Server limits the gateway inherits
These are Iggy server limits, not gateway settings. A Kafka client cannot act on any of them, so
diff --git a/gateways/kafka/docs/ACL_MAPPING.md b/gateways/kafka/docs/ACL_MAPPING.md
index 463158f7e..c552306f7 100644
--- a/gateways/kafka/docs/ACL_MAPPING.md
+++ b/gateways/kafka/docs/ACL_MAPPING.md
@@ -22,10 +22,8 @@ not rendered. See [What is not mapped](#what-is-not-mapped).
## Why this is worth doing before the data plane exists
-Authorization cannot be *enforced* yet. Produce and Fetch are stubs
-([#3535](https://github.com/apache/iggy/issues/3535),
-[#3536](https://github.com/apache/iggy/issues/3536)), so there is no operation to gate, and the
-verified identity is currently discarded.
+Authorization cannot be *enforced* yet. Produce and Fetch act as the bridge's Iggy user for every
+client, so no operation runs as the principal, and the verified identity is discarded.
`DescribeAcls` is the one authorization surface that needs neither. It is a read: take the
authenticated principal, take the permissions Iggy already holds for it, and answer in Kafka's
diff --git a/gateways/kafka/docs/AUTHENTICATION.md b/gateways/kafka/docs/AUTHENTICATION.md
index 43aaacfa5..67b27e29c 100644
--- a/gateways/kafka/docs/AUTHENTICATION.md
+++ b/gateways/kafka/docs/AUTHENTICATION.md
@@ -95,12 +95,10 @@ Two consequences worth naming rather than discovering:
traffic on that shard. Measured on debug builds, a single login takes about 10 ms and 32 concurrent
logins take about 75 ms each, with throughput flattening near 410 per second. The shape is the point,
latency climbing while throughput plateaus, not the absolute numbers. `MANUAL_TESTING.md` has the table.
-- **Nothing holds the verified session.** The connection that proved a credential drops it immediately,
- because no handler consumes an Iggy session yet. When Produce and Fetch need one
- ([#3535](https://github.com/apache/iggy/issues/3535),
- [#3536](https://github.com/apache/iggy/issues/3536)), the question of how a principal's data client is
- pooled becomes live, and it is a separate question from how its credentials are verified. Pooling clients
- is gateway-side code over the existing public SDK and needs no SDK change.
+- **Nothing holds the verified session.** The connection that proved a credential drops it at once.
+ Produce and Fetch act as the bridge's Iggy user for every client. A data client per principal needs
+ pooling, a separate question from how credentials are verified. Pooling is gateway-side code over the
+ public SDK and needs no SDK change.
If either consequence starts to hurt, the lever is below, and it is a smaller change than a credential
cache because it moves cost off the server without weakening what the gateway checks.
diff --git a/gateways/kafka/docs/BRIDGE_MAPPING.md b/gateways/kafka/docs/BRIDGE_MAPPING.md
index 8ae3cac77..2d5d2ecd1 100644
--- a/gateways/kafka/docs/BRIDGE_MAPPING.md
+++ b/gateways/kafka/docs/BRIDGE_MAPPING.md
@@ -88,7 +88,9 @@ The handler in [#3536](https://github.com/apache/iggy/issues/3536) owns the poli
the mapping refuses. Two shapes are on the table. Skipping serves the records around it and
leaves a gap, which Kafka consumers already tolerate on a compacted topic. Its cost is that a
message goes missing with no signal. Quarantining records the offset and surfaces a metric. Its
-cost is somewhere to keep the record. Neither is decided here.
+cost is somewhere to keep the record. Neither is decided here. Until one is, Fetch serves the
+records before such a message and answers `-1` at it, so the consumer stops at that offset.
+Fetch then reads that offset at most once per `max_wait_ms`.
### Timestamps
@@ -299,9 +301,9 @@ example for a request it classifies as a duplicate, in which case the response c
rather than a guessed offset; Kafka clients surface that as an unknown offset.
ListOffsets LATEST is the high watermark from `IggyBridge::high_watermarks`. EARLIEST has no
-server-side field today (`Partition` carries no log start offset), so it reads the first
-retained message instead, and the `(messages_count, current_offset) == (0, 0)` ambiguity
-documented on `high_watermarks` applies to both.
+server-side field today (`Partition` carries no log start offset), so it answers `0`. The
+`(messages_count, current_offset) == (0, 0)` ambiguity documented on `high_watermark` in
+`src/bridge/iggy_bridge/offsets.rs` applies to LATEST.
### Header order
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index e77194539..f649147e9 100644
--- a/gateways/kafka/docs/CONSUMER_GROUPS.md
+++ b/gateways/kafka/docs/CONSUMER_GROUPS.md
@@ -177,8 +177,8 @@ Once Metadata reports partitions, the next wall is `OffsetFetch` (9), which a co
`SyncGroup` for a non-empty assignment and which is not in scope
([#3542](https://github.com/apache/iggy/issues/3542)). An unlisted key closes the connection, so
that consumer would loop: coordinator connection closes, client marks the coordinator unknown,
-re-runs FindCoordinator, retries OffsetFetch, closes again. Fetch is a stub in any case, so nothing
-can be consumed until [#3535](https://github.com/apache/iggy/issues/3535)/#3542 land.
+re-runs FindCoordinator, retries OffsetFetch, closes again. Fetch reads records with a bridge, so a
+consumer that uses `assign()` outside a group can consume. A group member cannot until #3542 lands.
A dynamic consumer releases its partitions on close through `LeaveGroup`. A static one does not
send it (see [Static membership](#static-membership-is-accepted-not-honoured)), so a static
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index cce7af5f1..0edc6384c 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -336,7 +336,7 @@ Record kcat version and exact error strings in your test log. G1 passing is the
| -1 | UNKNOWN_SERVER_ERROR | Produce with a bridge: Iggy error with no closer code, or bad bridge login |
| 0 | NONE | Fetch top-level error field only (`ec=0` there does not mean per-partition success - see A6) |
| 3 | UNKNOWN_TOPIC_OR_PARTITION | Metadata stub, per topic. Produce with a bridge: missing topic or partition |
-| 6 | NOT_LEADER_OR_FOLLOWER | Produce/Fetch/ListOffsets stub (not stored). Produce with a bridge: Iggy unreachable, or the request budget ran out |
+| 6 | NOT_LEADER_OR_FOLLOWER | Produce/Fetch/ListOffsets stub (not stored). Produce with a bridge: Iggy unreachable, or the request budget ran out. Fetch with a bridge: Iggy unreachable, too slow, or loading the partition |
| 7 | REQUEST_TIMED_OUT | Produce with a bridge: deadline passed, or connection lost mid-send (may be stored) |
| 10 | MESSAGE_TOO_LARGE | Produce with a bridge: record, send or partition too large, even alone |
| 17 | INVALID_TOPIC_EXCEPTION | Produce with a bridge: bad topic name |
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index a331ca012..66bfe21d7 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -2,9 +2,9 @@
## Issue #3421 — in scope (this iteration)
-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. With a bridge, Produce writes to Iggy and ListOffsets reads offsets from it.
+A TCP listener on the Kafka wire port. It decodes requests, validates scoped API keys, versions and wire formats, and answers them. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy.
-**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)).
+**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). Fetch and ListOffsets answer 6 too. 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 records once you configure a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
| Deliverable | Status | Location |
| ------------- | -------- | ---------- |
@@ -46,7 +46,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 | With a bridge: one `send_messages` per partition. 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 |
+| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | With a bridge: shared topic probes, paged `poll_messages` per partition with records, wait up to `max_wait_ms` (max 18 s). Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
| 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+ |
| 10 | FindCoordinator | 0 | 4 | 0, 1, 2, 3, 4 | Answers "this gateway" for group keys; `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` (53) for the transaction key type, `INVALID_REQUEST` (42) for share; flexible encoding at v3+ |
@@ -132,8 +132,8 @@ untouched, at at-least-once delivery. See [`IDEMPOTENCE.md`](IDEMPOTENCE.md).
| Layer | #3421 | Description |
| ------- | ------- | ------------- |
| **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 12 keys; `bounds_guard.rs` pre-validates against unbounded allocation before handing a frame to the crate; stub responses except InitProducerId and the five consumer-group keys, and CreateTopics, Metadata, Produce and ListOffsets with a bridge |
-| **3 — Iggy bridge** | CreateTopics, Metadata, Produce and ListOffsets wired | `bridge/` module (connection, topic mapping, provisioning, high watermark, `topic_target` + `send_records`). CreateTopics ([#3538](https://github.com/apache/iggy/issues/3538)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)), 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 ([# [...]
+| **2 — Request/response codecs** | Partial | Decode/encode via the `kafka_protocol` crate (broker feature only) for 12 keys; `bounds_guard.rs` pre-validates against unbounded allocation before handing a frame to the crate; stub responses except InitProducerId and the five consumer-group keys, and Produce, Fetch, ListOffsets, Metadata and CreateTopics with a bridge |
+| **3 — Iggy bridge** | Produce, Fetch, ListOffsets, Metadata and CreateTopics wired | `bridge/` module (connection, topic mapping, provisioning, high watermark, `topic_target` + `send_records`, `probe` + `poll`). Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets ([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)) and CreateTopics ([#3538 [...]
---
@@ -274,8 +274,9 @@ below it are still open for the issues that build on top of it.
tracks no rolling low-watermark distinct from partition creation, so once a partition is
old enough for retention to purge its first segment, `0` names a log-start offset that no
longer exists - a real consumer with `auto.offset.reset=earliest` seeks into a hole. Not
- fixable client-side; needs the bridge to expose a real start offset. Harmless *today* only
- because Fetch (`#3536`) is still a stub - nothing yet reads at the offset this returns.
+ fixable client-side; needs the bridge to expose a real start offset. Fetch at an offset in
+ that hole reads from the oldest kept message, so the consumer skips the gap. If none is kept,
+ Fetch serves no records until the next write.
- Bridge fan-out is bounded independently of `bounds_guard`'s `MAX_REQUEST_ELEMENTS` (4,096,
still a pre-decode ceiling, not a usability one): topic entries sharing a name are deduped to
one `high_watermarks` call before any bridge work starts (a name repeated across request
@@ -288,6 +289,13 @@ below it are still open for the issues that build on top of it.
fixed ceiling, not a client-honored one), applied per topic rather than once around the whole
batch: a topic already resolved when the deadline arrives keeps its real answer, and only the
not-yet-started topics answer `REQUEST_TIMED_OUT`.
+- [x] Fetch → `poll_messages` ([#3536](https://github.com/apache/iggy/issues/3536)): with
+ `IGGY_KAFKA_BRIDGE_ENABLED=true`, shared topic probes, paged polls per partition with
+ records, and a wait of up to `max_wait_ms` (max 18 s) when there are none. Only codes the
+ Java consumer handles go out. See [README.md](../README.md#fetch-3536),
+ `src/protocol/handlers/fetch.rs`, `tests/fetch_real_bridge_tests.rs`.
+- [ ] Fetch: pick skip or quarantine for a stored message the gateway cannot map. Today the
+ consumer stops at it, and Fetch reads it at most once per `max_wait_ms`. See [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md#provenance).
- [ ] Real Metadata topology (brokers, partitions, leaders) backed by Iggy state
### `kafka-protocol` crate adoption — superseded, done differently
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index f8c2b6816..4153ec8c2 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -20,7 +20,7 @@ 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`, `list_offsets_real_bridge_tests` and `produce_real_bridge_tests`)
+### `iggy-server` binary (required for `bridge_iggy_integration_tests` and every `*_real_bridge_tests` suite)
No fixtures needed, but `iggy-server` has to be built *first* - these suites spawn it directly and
do not build it for you:
@@ -66,8 +66,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 |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
-| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker and a built `iggy-server` |
+| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker and a built `iggy-server` |
| [`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) |
+| [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, codes 3 and 43 | No (needs the `iggy-server` binary - see Prerequisites) |
+| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | 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`,
diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs
index d9560c1f9..d9fc6cb43 100644
--- a/gateways/kafka/src/bridge/error.rs
+++ b/gateways/kafka/src/bridge/error.rs
@@ -61,6 +61,9 @@ pub enum BridgeError {
partition: u32,
partitions_count: u32,
},
+ /// Iggy loads the partition, so its high watermark reads too low. Retriable: 6.
+ #[error("partition {partition} of topic '{topic}' is loading")]
+ PartitionLoading { topic: String, partition: u32 },
/// `ensure_topic` was asked to ensure a topic that already exists with a different partition
/// count. `ensure_topic`'s whole contract is "the topic has `partition_count` partitions
/// afterward" - silently keeping the old count and returning `Ok(())` would let two
@@ -105,6 +108,7 @@ impl BridgeError {
Self::Iggy(err) => iggy_error_to_kafka_code(err),
Self::Timeout | Self::SendLost(_) => ERROR_REQUEST_TIMED_OUT,
Self::PartitionOutOfRange { .. } => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ Self::PartitionLoading { .. } => ERROR_NOT_LEADER_OR_FOLLOWER,
Self::PartitionCountMismatch { .. } => ERROR_TOPIC_ALREADY_EXISTS,
Self::InvalidKafkaTopicName { .. } => ERROR_INVALID_TOPIC_EXCEPTION,
// Same code the wire-validation layer already uses for this exact condition - this
diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index eb630c943..3124e5a4a 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -17,5 +17,369 @@
//! Fetch-side bridge calls.
//!
-//! Empty until the Fetch handler lands. Separate from `produce.rs` so the two never edit one
-//! file.
+//! Separate from `produce.rs` so the two never edit one file.
+
+use std::sync::{Arc, Mutex, PoisonError};
+
+use iggy::prelude::{
+ Client, Consumer, IggyClient, IggyError, MessageClient, Partition, PolledMessages,
+ PollingStrategy, TopicClient,
+};
+use tokio::sync::{OnceCell, OwnedSemaphorePermit, Semaphore};
+use tokio::time::{Instant, timeout_at};
+
+use super::offsets::high_watermark;
+use super::{IggyBridge, TopicTarget, connect_client, in_slot, with_request_timeout};
+use crate::bridge::config::IggyBridgeConfig;
+use crate::bridge::error::BridgeError;
+
+/// Fetch reads that poll at once. Each has its own Iggy client, so a poll that Iggy holds up
+/// stalls only its own slot.
+const FETCH_SLOTS: usize = 4;
+
+/// The Fetch read slots, with a client for each.
+pub(super) struct FetchPool {
+ permits: Arc<Semaphore>,
+ /// The clients of the free slots.
+ idle: Mutex<Vec<FetchClient>>,
+ /// Every slot's client, for `close`.
+ all: Vec<FetchClient>,
+}
+
+impl FetchPool {
+ pub(super) fn new() -> Self {
+ let all: Vec<FetchClient> = (0..FETCH_SLOTS).map(|_| FetchClient::default()).collect();
+ Self {
+ permits: Arc::new(Semaphore::new(FETCH_SLOTS)),
+ idle: Mutex::new(all.clone()),
+ all,
+ }
+ }
+
+ /// A free slot. `None` if none frees by `deadline`.
+ async fn take(self: &Arc<Self>, deadline: Instant) -> Option<FetchSlot> {
+ let permit = timeout_at(deadline, Arc::clone(&self.permits).acquire_owned())
+ .await
+ .ok()?
+ .ok()?;
+ let client = self
+ .idle
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .pop()?;
+ Some(FetchSlot {
+ client,
+ pool: Arc::clone(self),
+ _permit: permit,
+ })
+ }
+
+ /// Shuts down each client that connected.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ for client in &self.all {
+ if let Some(client) = client.0.get() {
+ with_request_timeout(client.shutdown()).await?;
+ }
+ }
+ Ok(())
+ }
+}
+
+/// The Iggy client of one read slot. It connects at the slot's first poll, so a bridge that never
+/// fetches opens no extra connection.
+#[derive(Clone, Default)]
+struct FetchClient(Arc<OnceCell<IggyClient>>);
+
+impl FetchClient {
+ async fn connected(&self, config: &IggyBridgeConfig) -> Result<&IggyClient, BridgeError> {
+ self.0.get_or_try_init(|| connect_client(config)).await
+ }
+}
+
+/// One Fetch read slot. Dropping it frees the slot.
+pub struct FetchSlot {
+ client: FetchClient,
+ pool: Arc<FetchPool>,
+ _permit: OwnedSemaphorePermit,
+}
+
+impl Drop for FetchSlot {
+ fn drop(&mut self) {
+ // Before the permit goes, so the next holder finds a client.
+ self.pool
+ .idle
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .push(self.client.clone());
+ }
+}
+
+/// One partition as a Fetch sees it before it reads.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PartitionProbe {
+ /// Same value `high_watermarks` reports.
+ pub high_watermark: u64,
+ /// Stored bytes per retained message. 0 when the partition retains none.
+ pub average_size: u64,
+ /// 0 when retention removed every message, or when the server has no stats for the partition.
+ pub messages_count: u64,
+}
+
+impl PartitionProbe {
+ /// Iggy counts the messages of a partition it loads before it restores the offset, so the high
+ /// watermark reads too low until the load ends.
+ ///
+ /// A commit between Iggy's reads of the offset and the count looks the same for one probe.
+ /// [`IggyBridge::probe`] reads twice to tell them apart.
+ #[must_use]
+ pub const fn is_loading(&self) -> bool {
+ self.messages_count > self.high_watermark
+ }
+}
+
+impl From<&Partition> for PartitionProbe {
+ fn from(partition: &Partition) -> Self {
+ Self {
+ high_watermark: high_watermark(partition),
+ average_size: partition
+ .size
+ .as_bytes_u64()
+ .checked_div(partition.messages_count)
+ .unwrap_or(0),
+ messages_count: partition.messages_count,
+ }
+ }
+}
+
+/// Every partition of one topic, by id.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct TopicProbe {
+ partitions: Vec<(u32, PartitionProbe)>,
+}
+
+impl TopicProbe {
+ /// The probe of partition `id`, if the topic has it.
+ #[must_use]
+ pub fn get(&self, id: u32) -> Option<PartitionProbe> {
+ self.partitions
+ .binary_search_by_key(&id, |&(found, _)| found)
+ .ok()
+ .map(|found| self.partitions[found].1)
+ }
+
+ #[must_use]
+ pub fn partitions_count(&self) -> u32 {
+ u32::try_from(self.partitions.len()).unwrap_or(u32::MAX)
+ }
+}
+
+impl FromIterator<(u32, PartitionProbe)> for TopicProbe {
+ fn from_iter<I: IntoIterator<Item = (u32, PartitionProbe)>>(partitions: I) -> Self {
+ let mut partitions: Vec<_> = partitions.into_iter().collect();
+ partitions.sort_unstable_by_key(|&(id, _)| id);
+ Self { partitions }
+ }
+}
+
+impl IggyBridge {
+ /// A Fetch read slot, with its own client. `None` if none frees by `deadline`.
+ pub async fn fetch_slot(&self, deadline: Instant) -> Option<FetchSlot> {
+ self.fetch_pool.take(deadline).await
+ }
+
+ /// Probes every partition of the topic `kafka_topic` maps to, with one `get_topic` call, or
+ /// two when a partition reads as loading.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::InvalidKafkaTopicName`] if the name fails Kafka's rules.
+ /// [`BridgeError::Iggy`] if the stream or topic is missing, or the call fails.
+ /// [`BridgeError::Timeout`] past `REQUEST_TIMEOUT`.
+ pub async fn probe(&self, kafka_topic: &str) -> Result<TopicProbe, BridgeError> {
+ let target = self.topic_target(kafka_topic)?;
+ let probe = self.probe_target(&target).await?;
+ // A load lasts. A commit that raced the read does not.
+ if probe
+ .partitions
+ .iter()
+ .any(|(_, partition)| partition.is_loading())
+ {
+ return self.probe_target(&target).await;
+ }
+ Ok(probe)
+ }
+
+ async fn probe_target(&self, target: &TopicTarget) -> Result<TopicProbe, BridgeError> {
+ let details =
+ with_request_timeout(self.client.get_topic(&target.stream_id, &target.topic_id))
+ .await?
+ // A missing stream reads as `None` too. Both answer the same Kafka code.
+ .ok_or_else(|| {
+ BridgeError::Iggy(IggyError::TopicNameNotFound(
+ target.topic_id.to_string(),
+ target.stream_id.to_string(),
+ ))
+ })?;
+ Ok(details
+ .partitions
+ .iter()
+ .map(|partition| (partition.id, PartitionProbe::from(partition)))
+ .collect())
+ }
+
+ /// Reads up to `count` messages of `partition`, from `offset` on, with the client of `slot`.
+ /// Hands the slot back with the result. `None` if no result comes by `deadline`.
+ ///
+ /// A plain consumer without auto commit, so Iggy stores no offset. Always an explicit
+ /// offset, never `Next` (see `docs/OFFSET_STORAGE.md`).
+ ///
+ /// The errors are those of [`Self::probe`]. A partition the topic lacks is
+ /// [`BridgeError::Iggy`] here. The poll stops at `deadline` too, so a result that comes then
+ /// is [`BridgeError::Timeout`].
+ pub async fn poll(
+ self: &Arc<Self>,
+ slot: FetchSlot,
+ kafka_topic: &str,
+ partition: u32,
+ offset: u64,
+ count: u32,
+ deadline: Instant,
+ ) -> Option<(Result<PolledMessages, BridgeError>, FetchSlot)> {
+ let (bridge, client, topic) = (
+ Arc::clone(self),
+ slot.client.clone(),
+ kafka_topic.to_owned(),
+ );
+ let poll = async move {
+ bridge
+ .poll_with(&client, &topic, partition, offset, count, deadline)
+ .await
+ };
+ in_slot(slot, deadline, poll).await
+ }
+
+ async fn poll_with(
+ &self,
+ client: &FetchClient,
+ kafka_topic: &str,
+ partition: u32,
+ offset: u64,
+ count: u32,
+ deadline: Instant,
+ ) -> Result<PolledMessages, BridgeError> {
+ let target = self.topic_target(kafka_topic)?;
+ let (consumer, strategy) = (Consumer::default(), PollingStrategy::offset(offset));
+ let poll = async {
+ client
+ .connected(&self.config)
+ .await?
+ .poll_messages(
+ &target.stream_id,
+ &target.topic_id,
+ Some(partition),
+ &consumer,
+ &strategy,
+ count,
+ false,
+ )
+ .await
+ .map_err(BridgeError::Iggy)
+ };
+ // The SDK replays a poll that Iggy refuses for up to 30 s. Stop with the request, so the
+ // slot frees.
+ match timeout_at(deadline, poll).await {
+ Ok(polled) => polled,
+ Err(_elapsed) => Err(BridgeError::Timeout),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use iggy::prelude::{IggyByteSize, IggyTimestamp};
+
+ use super::*;
+
+ #[tokio::test]
+ async fn given_every_slot_taken_when_one_is_dropped_should_hand_its_client_to_the_next_read() {
+ let pool = Arc::new(FetchPool::new());
+ let soon = || Instant::now() + Duration::from_millis(20);
+ let mut taken = Vec::new();
+ for _ in 0..FETCH_SLOTS {
+ taken.push(pool.take(soon()).await.expect("a free slot"));
+ }
+ for (index, slot) in taken.iter().enumerate() {
+ for other in &taken[index + 1..] {
+ assert!(
+ !Arc::ptr_eq(&slot.client.0, &other.client.0),
+ "each slot has its own client"
+ );
+ }
+ }
+ assert!(pool.take(soon()).await.is_none(), "every slot is taken");
+
+ let freed = taken.pop().expect("a slot to free");
+ let client = freed.client.clone();
+ drop(freed);
+ let next = pool.take(soon()).await.expect("the freed slot");
+ assert!(
+ Arc::ptr_eq(&next.client.0, &client.0),
+ "with the freed client"
+ );
+ drop((next, taken));
+ }
+
+ fn partition(current_offset: u64, messages_count: u64, size: u64) -> Partition {
+ Partition {
+ id: 0,
+ created_at: IggyTimestamp::zero(),
+ segments_count: 1,
+ current_offset,
+ size: IggyByteSize::from(size),
+ messages_count,
+ }
+ }
+
+ #[test]
+ fn given_stored_messages_when_probed_should_give_watermark_and_average_size() {
+ let probe = PartitionProbe::from(&partition(9, 10, 1000));
+ assert_eq!(probe.high_watermark, 10);
+ assert_eq!(probe.average_size, 100);
+ assert_eq!(probe.messages_count, 10);
+ }
+
+ #[test]
+ fn given_an_empty_partition_when_probed_should_give_zero_for_both() {
+ let probe = PartitionProbe::from(&partition(0, 0, 0));
+ assert_eq!(probe.high_watermark, 0);
+ assert_eq!(probe.average_size, 0);
+ }
+
+ #[test]
+ fn given_every_message_expired_when_probed_should_keep_the_watermark() {
+ // Retention lowers the count, not the offset.
+ let probe = PartitionProbe::from(&partition(41, 0, 0));
+ assert_eq!(probe.high_watermark, 42);
+ assert_eq!(probe.average_size, 0, "no retained message to measure");
+ assert_eq!(probe.messages_count, 0);
+ }
+
+ #[test]
+ fn given_partitions_in_any_order_when_looked_up_should_find_each_by_id() {
+ let probe = |high_watermark| PartitionProbe {
+ high_watermark,
+ average_size: 1,
+ messages_count: high_watermark,
+ };
+ let topic: TopicProbe = [(2, probe(20)), (0, probe(0)), (5, probe(50))]
+ .into_iter()
+ .collect();
+ assert_eq!(topic.get(0), Some(probe(0)));
+ assert_eq!(topic.get(2), Some(probe(20)));
+ assert_eq!(topic.get(5), Some(probe(50)));
+ assert_eq!(topic.get(3), None);
+ assert_eq!(topic.partitions_count(), 3);
+ }
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index 34328f02e..367b9e2b8 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -22,7 +22,8 @@ use std::time::Duration;
use iggy::prelude::{
AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder, IggyError,
};
-use tokio::sync::Semaphore;
+use tokio::sync::{Semaphore, oneshot};
+use tokio::time::{Instant, timeout_at};
use tracing::info;
use crate::bridge::config::IggyBridgeConfig;
@@ -34,6 +35,8 @@ mod offsets;
mod produce;
mod topics;
+use fetch::FetchPool;
+pub use fetch::{FetchSlot, PartitionProbe, TopicProbe};
pub use topics::{KafkaTopicMetadata, TopicCreationOutcome};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
@@ -74,15 +77,39 @@ async fn with_request_timeout<T>(
.map_err(BridgeError::Iggy)
}
-/// Owns one connected `IggyClient` and resolves Kafka topics against it.
+/// How long a send may hold its slot: the SDK read deadline (30 s) plus a reconnect (15 s).
+const SLOT_LIMIT: Duration = Duration::from_secs(45);
+
+/// Runs `call` in a task that holds `slot` until the call ends, then hands both back. `None` once
+/// `deadline` passes.
+///
+/// The SDK cannot cancel a call, so a call given up on keeps its slot. The next call then waits for
+/// the slot, not in the SDK queue.
+async fn in_slot<S: Send + 'static, T: Send + 'static>(
+ slot: S,
+ deadline: Instant,
+ call: impl Future<Output = T> + Send + 'static,
+) -> Option<(T, S)> {
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let done = call.await;
+ // The caller may have stopped waiting. The slot then goes back here.
+ let _ = sender.send((done, slot));
+ });
+ timeout_at(deadline, receiver).await.ok()?.ok()
+}
+
+/// Owns the connected `IggyClient`s and resolves Kafka topics against them.
///
-/// One lockstep client serves every Kafka connection, so Iggy calls run one at a time. A pool is
-/// a TODO in `docs/SCOPE.md`.
+/// One lockstep client serves every Kafka connection, so its Iggy calls run one at a time. Fetch
+/// polls use their own clients, one per read slot. A Produce pool is a TODO in `docs/SCOPE.md`.
pub struct IggyBridge {
client: Arc<IggyClient>,
config: IggyBridgeConfig,
/// One Produce send inside the SDK at a time. See `send_records`.
send_slot: Arc<Semaphore>,
+ /// Fetch read slots, each with its own client. See `fetch_slot`.
+ fetch_pool: Arc<FetchPool>,
}
/// The Iggy stream and topic one Kafka topic maps to. Resolve once per topic, use many times.
@@ -111,28 +138,14 @@ impl IggyBridge {
/// to a wire response, never panic or unwrap, since an unreachable Iggy backend is an
/// expected runtime condition, not a bug.
pub async fn connect(config: IggyBridgeConfig) -> Result<Self, BridgeError> {
- if config.address.trim().is_empty() {
- return Err(BridgeError::InvalidConfig(
- "Iggy address must not be empty".to_string(),
- ));
- }
-
- let credentials =
- Credentials::UsernamePassword(config.username.clone(), config.password.clone());
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(config.address.clone())
- .with_auto_sign_in(AutoLogin::Enabled(credentials))
- .with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
- .build()
- .map_err(BridgeError::Iggy)?;
- with_request_timeout(client.connect()).await?;
+ let client = connect_client(&config).await?;
info!("Iggy bridge connected to {}", config.address);
Ok(Self {
client: Arc::new(client),
config,
send_slot: Arc::new(Semaphore::new(1)),
+ fetch_pool: Arc::new(FetchPool::new()),
})
}
@@ -175,6 +188,28 @@ impl IggyBridge {
/// [`BridgeError::Iggy`] if the underlying client reports a shutdown failure (e.g. the socket
/// was already in a state that rejects a clean shutdown).
pub async fn close(self) -> Result<(), BridgeError> {
+ self.fetch_pool.close().await?;
with_request_timeout(self.client.shutdown()).await
}
}
+
+/// A client for `config`, connected and signed in. See [`IggyBridge::connect`].
+async fn connect_client(config: &IggyBridgeConfig) -> Result<IggyClient, BridgeError> {
+ if config.address.trim().is_empty() {
+ return Err(BridgeError::InvalidConfig(
+ "Iggy address must not be empty".to_string(),
+ ));
+ }
+
+ let credentials =
+ Credentials::UsernamePassword(config.username.clone(), config.password.clone());
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address(config.address.clone())
+ .with_auto_sign_in(AutoLogin::Enabled(credentials))
+ .with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
+ .build()
+ .map_err(BridgeError::Iggy)?;
+ with_request_timeout(client.connect()).await?;
+ Ok(client)
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs
index 6a1a30650..9a846a60a 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs
@@ -17,132 +17,48 @@
//! Offset and high-watermark lookups.
-use iggy::prelude::{IggyError, TopicClient};
+use iggy::prelude::Partition;
-use super::{IggyBridge, with_request_timeout};
+use super::IggyBridge;
use crate::bridge::error::BridgeError;
impl IggyBridge {
- /// Returns the high watermark (one past the highest *committed* offset - see
- /// `Partition::offset_frontier`'s own definition) for every partition in `partitions` of the
- /// Iggy topic `kafka_topic` maps to, in one round trip.
+ /// The high watermark of each of `partitions`, in the same order, from one [`Self::probe`].
///
- /// Takes `kafka_topic`, not raw Iggy stream/topic names, and resolves it through the same
- /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping) `ensure_stream_and_topic`
- /// uses - a caller (a future `ListOffsets` handler) only ever has the Kafka-side name, and a
- /// topic with a mapping override would silently query the wrong Iggy resource if this took
- /// Iggy-space names directly instead.
- ///
- /// One call to `get_topic`, not one per partition: a single Kafka `ListOffsets` request asks
- /// about many partitions of one topic at once (`ListOffsetsRequest.json`'s
- /// `Topics[] -> Partitions[]`), and `get_topic`'s decode already rebuilds and sorts the whole
- /// partition vector regardless of how many of them the caller wants
- /// (`wire_conversions.rs`) - a naive one-call-per-partition wrapper around a
- /// single-partition method would turn one Kafka request into N round trips and N times the
- /// decode work. Returns results in the same order as `partitions`.
- ///
- /// `i64`, not `u64`: the `ListOffsets` response field this exists to fill is `int64`
- /// (`ListOffsetsResponse.json`), so a future handler needs no cast at the wire boundary - the
- /// lossy conversion happens once, here (`Partition::current_offset` is `u64`; converted via
- /// `i64::try_from`, saturating to `i64::MAX` on the practically-unreachable overflow case
- /// rather than panicking or wrapping).
- ///
- /// `Partition::current_offset` is the offset of the *last written* message, not "next offset
- /// to produce" - confirmed against a live server (3 produced messages read back
- /// `current_offset == 2`). An empty partition has no last-written offset at all, so this
- /// needs a dedicated empty case rather than inferring it from `current_offset == 0` (also a
- /// fresh partition's default value, indistinguishable from "one message at offset 0").
- ///
- /// That empty case is `messages_count == 0 && current_offset == 0`, not `messages_count == 0`
- /// alone: retention cleanup decrements `messages_count`
- /// (`iggy_partition.rs::decrement_messages_count`) without rewinding `current_offset`, so a
- /// fully-purged but previously-produced-to partition would otherwise read as empty. This is
- /// still not exact: a partition whose *only* message, at offset 0, gets trimmed by the same
- /// retention pass reports `(messages_count, current_offset) == (0, 0)` too, so its watermark
- /// would read `1` then drop back to `0` - a real Kafka watermark never moves backward. Narrow
- /// (needs retention to fully empty a partition that only ever held one message) and not
- /// fixable client-side (`PartitionResponse` carries no frontier field to disambiguate it), but
- /// worth having on the record rather than only implicitly true of the check below.
- ///
- /// Not the same value as `Partition::mint_frontier` (the offset the *next* mint will take):
- /// after a crash-recovery reservation, the append point can sit above the committed frontier
- /// by the reservation's lease block, and neither value is on the wire to tell them apart. This
- /// is still correct for `ListOffsets` LATEST (defined in terms of the committed offset), but a
- /// future Produce handler must not use it to predict a base offset for a write in flight.
- ///
- /// Known remaining gap, not fixable client-side: `(0, 0)` is also what a stats-registry MISS
- /// reports (`responses.rs`'s `PartitionResponse` builder), indistinguishable on the wire from
- /// a genuinely empty partition. Narrow and self-healing - `partition_reconciler.rs`'s
- /// `settle_partition_stats` opens this only on the teardown-for-rebuild path, closing once the
- /// rebuild completes; deletes never open it. The real answer (`Partition::offset_frontier`)
- /// stays server-side and isn't in this response. The check matches the server's own
- /// `PartitionState::store_offset_range_error` condition, so it's not an invented heuristic - a
- /// future `ListOffsets` (`#3537`) built on this inherits the same blind spot.
+ /// `i64`, as `ListOffsets` sends it. A partition the topic lacks, or one Iggy loads, fails
+ /// alone.
///
/// # Errors
///
- /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
- /// topic-naming rules. Returns [`BridgeError::Timeout`] if the call takes longer than
- /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] if the mapped stream doesn't exist (see
- /// the note above on which resource is actually missing). These are call-level failures -
- /// nothing about any individual partition could be resolved. A single out-of-range
- /// `partition` does not fail the whole call: it is reported per-partition (see the return
- /// type), so the partitions that did resolve are never discarded to report one that didn't.
+ /// As [`Self::probe`].
pub async fn high_watermarks(
&self,
kafka_topic: &str,
partitions: &[u32],
) -> Result<Vec<(u32, Result<i64, BridgeError>)>, BridgeError> {
- let target = self.topic_target(kafka_topic)?;
- let details =
- with_request_timeout(self.client.get_topic(&target.stream_id, &target.topic_id))
- .await?
- .ok_or_else(|| {
- // A missing *stream* also makes get_topic return Ok(None) (responses.rs), so this
- // reports TopicNameNotFound even when the stream is what's actually gone - both
- // map to the same Kafka wire code either way, so only the log text is affected.
- BridgeError::Iggy(IggyError::TopicNameNotFound(
- target.topic_id.to_string(),
- target.stream_id.to_string(),
- ))
- })?;
-
+ let topic = self.probe(kafka_topic).await?;
Ok(partitions
.iter()
.map(|&partition| {
- // `TryFrom<GetTopicResponse> for TopicDetails` (wire_conversions.rs) sorts
- // `partitions` by `id` on every decode, so a binary search is correct here, not
- // just faster than a linear scan - for a 1000-partition topic, the difference is
- // O(log n) vs O(n) probes per requested partition.
- let watermark = details
- .partitions
- .binary_search_by_key(&partition, |p| p.id)
- .map(|index| &details.partitions[index])
- .map_err(|_| BridgeError::PartitionOutOfRange {
- // The Kafka-side name the caller asked about. A mapping override
- // would otherwise quote an Iggy name the Kafka client never heard of.
+ let watermark = match topic.get(partition) {
+ None => Err(BridgeError::PartitionOutOfRange {
+ topic: kafka_topic.to_string(),
+ partition,
+ partitions_count: topic.partitions_count(),
+ }),
+ // A seek to that watermark lands short, so the consumer reads again.
+ Some(probe) if probe.is_loading() => Err(BridgeError::PartitionLoading {
topic: kafka_topic.to_string(),
partition,
- partitions_count: details.partitions_count,
- })
- .map(|partition_details| {
- if partition_details.messages_count == 0
- && partition_details.current_offset == 0
- {
- 0
- } else {
- partition_details.current_offset.saturating_add(1)
- }
- })
- .map(|watermark| i64::try_from(watermark).unwrap_or(i64::MAX));
+ }),
+ Some(probe) => Ok(i64::try_from(probe.high_watermark).unwrap_or(i64::MAX)),
+ };
(partition, watermark)
})
.collect())
}
- /// Convenience wrapper around [`Self::high_watermarks`] for a single partition. See that
- /// method's doc for the full semantics - callers wanting more than one partition of the same
- /// topic should call it directly instead of this in a loop, to get its one-round-trip batching.
+ /// [`Self::high_watermarks`] for one partition.
///
/// # Errors
///
@@ -169,3 +85,19 @@ impl IggyBridge {
watermark
}
}
+
+/// One past the last committed offset of `partition`, or 0 when it is empty.
+///
+/// Fetch and `ListOffsets` both use this, so a consumer that seeks to LATEST fetches in range.
+///
+/// - Empty needs both counters: retention lowers `messages_count`, never `current_offset`.
+/// - Known gap: `(0, 0)` also comes back when the only message, at 0, expired, or when the server
+/// has no stats for the partition yet.
+/// - Not the offset of the next write: the server can reserve offsets past it.
+pub(super) const fn high_watermark(partition: &Partition) -> u64 {
+ if partition.messages_count == 0 && partition.current_offset == 0 {
+ 0
+ } else {
+ partition.current_offset.saturating_add(1)
+ }
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/produce.rs b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
index fe261887b..5a7c4689e 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/produce.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
@@ -21,18 +21,14 @@
use std::future::Future;
use std::sync::Arc;
-use std::time::Duration;
use iggy::prelude::{IggyError, IggyMessage, MessageClient, Partitioning};
-use tokio::sync::{Semaphore, oneshot};
+use tokio::sync::Semaphore;
use tokio::time::{Instant, timeout, timeout_at};
-use super::{IggyBridge, TopicTarget};
+use super::{IggyBridge, SLOT_LIMIT, TopicTarget, in_slot};
use crate::bridge::error::BridgeError;
-/// How long a send may hold the slot: the SDK read deadline (30 s) plus a reconnect (15 s).
-const SEND_LIMIT: Duration = Duration::from_secs(45);
-
/// Frame bytes besides the messages: request header, batch header and ids.
const SEND_FRAME_ROOM: u64 = 4 * 1024;
@@ -51,7 +47,7 @@ impl IggyBridge {
/// `Ok(None)`: committed, but the server named no offset (a duplicate request, or a journal
/// entry that is gone). Do not retry it.
///
- /// A stuck send holds the slot up to `SEND_LIMIT`. Other sends answer 7 meanwhile.
+ /// A stuck send holds the slot up to `SLOT_LIMIT`. Other sends answer 7 meanwhile.
///
/// # Errors
///
@@ -96,20 +92,16 @@ async fn send_in_slot<T: Send + 'static>(
Ok(Ok(permit)) if Instant::now() < deadline => permit,
_ => return Err(BridgeError::Timeout),
};
- let (sender, receiver) = oneshot::channel();
- tokio::spawn(async move {
- let result = match timeout(SEND_LIMIT, send).await {
+ let send = async move {
+ match timeout(SLOT_LIMIT, send).await {
Ok(result) => result.map_err(send_error),
Err(_elapsed) => Err(BridgeError::Timeout),
- };
- drop(permit);
- // The caller may have stopped waiting.
- let _ = sender.send(result);
- });
- match timeout_at(deadline, receiver).await {
- Ok(Ok(result)) => result,
- _ => Err(BridgeError::Timeout),
- }
+ }
+ };
+ // The permit comes back only to be dropped. A send that outlives the caller frees it in its task.
+ in_slot(permit, deadline, send)
+ .await
+ .map_or(Err(BridgeError::Timeout), |(result, _permit)| result)
}
/// The SDK does not replay a send after these, so it may have landed.
@@ -126,6 +118,9 @@ const fn send_error(error: IggyError) -> BridgeError {
#[cfg(test)]
mod tests {
use std::sync::atomic::{AtomicBool, Ordering};
+ use std::time::Duration;
+
+ use tokio::sync::oneshot;
use super::*;
diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs
index fc31934fb..548146564 100644
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@ -18,8 +18,8 @@
//! Iggy SDK integration layer.
//!
//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning and
-//! high-watermark lookups, and translates Iggy errors to Kafka wire error codes. `ListOffsets`
-//! and Produce call it. Fetch does not yet (`#3536`).
+//! high-watermark lookups, and translates Iggy errors to Kafka wire error codes. `ListOffsets`,
+//! Produce, Fetch, Metadata and `CreateTopics` call it.
pub mod config;
pub mod error;
@@ -28,5 +28,8 @@ pub mod topic_map;
pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig};
pub use error::BridgeError;
-pub use iggy_bridge::{IggyBridge, KafkaTopicMetadata, TopicCreationOutcome, TopicTarget};
+pub use iggy_bridge::{
+ FetchSlot, IggyBridge, KafkaTopicMetadata, PartitionProbe, TopicCreationOutcome, TopicProbe,
+ TopicTarget,
+};
pub use topic_map::{TopicMapping, TopicOverride};
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index 05eb9ab7b..2c2c8f1be 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -15,7 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
use bytes::Bytes;
use kafka_protocol::error::ResponseError;
@@ -41,6 +42,7 @@ use crate::protocol::handlers::{
init_producer_id, join_group, leave_group, list_offsets, metadata, produce, respond_or_close,
sync_group,
};
+use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
SaslMechanism, encode_sasl_authenticate_response, encode_sasl_handshake_response,
};
@@ -67,6 +69,8 @@ pub const DEFAULT_KAFKA_PORT: u16 = 9093;
/// uses it for an `IggyError` with no closer Kafka analogue.
pub const ERROR_UNKNOWN_SERVER_ERROR: i16 = ResponseError::UnknownServerError.code();
pub const ERROR_NONE: i16 = 0;
+/// Fetch: the offset is negative or past the high watermark. The client resets its position.
+pub const ERROR_OFFSET_OUT_OF_RANGE: i16 = ResponseError::OffsetOutOfRange.code();
pub const ERROR_UNKNOWN_TOPIC_OR_PARTITION: i16 = ResponseError::UnknownTopicOrPartition.code();
/// Retriable, nothing written. The Produce stub, and a partition refused because the request
/// budget ran out.
@@ -163,6 +167,9 @@ pub const ERROR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED: i16 = 53;
/// unknown user the same way, and distinguishing them here would reintroduce a user-enumeration
/// oracle.
pub const ERROR_SASL_AUTHENTICATION_FAILED: i16 = ResponseError::SaslAuthenticationFailed.code();
+/// Fetch: the request continues a session. This gateway opens none, so the client starts over
+/// with a full request.
+pub const ERROR_FETCH_SESSION_ID_NOT_FOUND: i16 = ResponseError::FetchSessionIdNotFound.code();
/// Produce: zstd before v7.
pub const ERROR_UNSUPPORTED_COMPRESSION_TYPE: i16 =
ResponseError::UnsupportedCompressionType.code();
@@ -179,6 +186,13 @@ pub const ERROR_FENCED_INSTANCE_ID: i16 = 82;
/// 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 = ResponseError::InvalidRecord.code();
+/// Kafka's -1 for an offset or a timestamp that is not known.
+pub(crate) const UNKNOWN_OFFSET: i64 = -1;
+
+/// Ceiling on the bridge work of one request, wait included. A clock, not a partition cap, so a
+/// wide topic still works. Above one bridge call's 15 s timeout. Fits the 25 s shutdown drain.
+pub(crate) const REQUEST_DEADLINE: Duration = Duration::from_secs(20);
+
/// Result of handling one Kafka request body.
#[derive(Debug)]
pub enum HandleOutcome {
@@ -276,8 +290,8 @@ pub fn supported_api_ranges() -> &'static [ApiVersionRange] {
/// `bridge` is `None` until `IGGY_KAFKA_BRIDGE_ENABLED` turns it on. A handler that finds `None`
/// answers with its stub, so APIs can be wired one at a time.
///
-/// One `IggyBridge` is one `IggyClient` and its TCP transport is lockstep, so Kafka connections
-/// serialize behind whichever Iggy request is in flight. The `Arc` does not change that. See the
+/// Every Iggy call but a Fetch poll goes through one lockstep `IggyClient`, so Kafka connections
+/// serialize behind whichever of those calls is in flight. The `Arc` does not change that. See the
/// README's "Concurrency ceiling".
pub struct GatewayState {
pub broker: BrokerAdvertise,
@@ -294,6 +308,12 @@ pub struct GatewayState {
/// Consumer group membership. Process-wide and independent of the bridge: a member outlives
/// the connection that created it, and group coordination needs no Iggy call.
pub groups: GroupCoordinator,
+ /// Partitions that answered -1 to Fetch, per Kafka topic.
+ pub(crate) stuck_offsets: fetch::StuckOffsets,
+ /// Per Kafka topic and partition, how long Fetch finds it not ready.
+ pub(crate) not_ready: Mutex<fetch::Spells>,
+ /// Topic probes that every Fetch shares.
+ pub(crate) probe_board: ProbeBoard,
}
/// Each holds one decoded partition at a time, so about 160 MB at the default 8 MiB frame. Sends
@@ -318,6 +338,9 @@ impl GatewayState {
producer_ids: ProducerIdAllocator::new(instance_id),
produce_slots: Semaphore::const_new(PRODUCE_SLOTS),
groups,
+ stuck_offsets: fetch::StuckOffsets::default(),
+ not_ready: Mutex::default(),
+ probe_board: ProbeBoard::default(),
}
}
diff --git a/gateways/kafka/src/protocol/handlers/fetch.rs b/gateways/kafka/src/protocol/handlers/fetch.rs
index c1e5e685f..3e50ad217 100644
--- a/gateways/kafka/src/protocol/handlers/fetch.rs
+++ b/gateways/kafka/src/protocol/handlers/fetch.rs
@@ -17,17 +17,43 @@
//! Fetch (API key 1).
+use std::collections::{HashMap, HashSet};
+use std::future::{Future, poll_fn};
+use std::sync::{Arc, Mutex, PoisonError};
+use std::task::Poll;
+use std::time::Duration;
+
use bytes::Bytes;
+use iggy::prelude::{
+ IGGY_MESSAGE_HEADER_SIZE, IggyMessage, MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE, PolledMessages,
+};
+use kafka_protocol::indexmap::IndexSet;
use kafka_protocol::messages::fetch_response::{FetchableTopicResponse, PartitionData};
use kafka_protocol::messages::{FetchRequest, FetchResponse};
+use kafka_protocol::protocol::Encodable;
+use kafka_protocol::records::Record;
+use tokio::sync::watch;
+use tokio::time::{Instant, sleep_until};
+use crate::bridge::topic_map::validate_kafka_topic_name;
+use crate::bridge::{BridgeError, FetchSlot, IggyBridge, PartitionProbe};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_FETCH, ApiVersionRange, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, GatewayState,
- HandleOutcome,
+ API_KEY_FETCH, ApiVersionRange, ERROR_FETCH_SESSION_ID_NOT_FOUND,
+ ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER,
+ ERROR_OFFSET_OUT_OF_RANGE, ERROR_REQUEST_TIMED_OUT, ERROR_TOPIC_AUTHORIZATION_FAILED,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome,
+ REQUEST_DEADLINE, UNKNOWN_OFFSET,
};
use crate::protocol::bounds_guard::validate_fetch_shape;
-use crate::protocol::handlers::{decode_guarded, encode_message, handle_versioned_request};
+use crate::protocol::handlers::{
+ CODEC_OFF_WORKER_BYTES, COPY_OFF_WORKER_BYTES, decode_guarded, decode_request, encode_message,
+ off_worker, respond_or_close,
+};
+use crate::protocol::probe_board::{PROBE_INTERVAL, Snapshot, TopicProbed, recent};
+use crate::records::{
+ BATCH_HEADER_BYTES, RecordCodecError, encode_batch, from_iggy, record_size_bound,
+};
pub const RANGE: ApiVersionRange = ApiVersionRange {
api_key: API_KEY_FETCH,
@@ -35,24 +61,79 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
max_version: 12,
};
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until a handler awaits the bridge"
-)]
+/// Longest wait. Leaves 2 s for the last probe and read, so a long wait does not end in 6.
+const MAX_WAIT: Duration = REQUEST_DEADLINE.saturating_sub(Duration::from_secs(2));
+/// Most topics one request loads a new probe for, per round, so a request that names many costs
+/// Iggy this many `get_topic` calls per round at most.
+const MAX_ROUND_TOPICS: usize = 100;
+/// How long a partition with no offset past 0 answers 6 instead of 1. Iggy reports that shape
+/// while it loads or fences a partition, so a 1 then resets a consumer that is in range.
+///
+/// A purge reads the same. Seek consumers to 0 after one: writes that pass a consumer's old offset
+/// make it skip the records below.
+const RELOAD_GRACE: Duration = Duration::from_secs(30);
+
+/// Most messages one partition read takes from Iggy, over all its polls.
+const MAX_POLL_COUNT: u32 = 1000;
+/// Most messages one poll asks for. Iggy caps a poll by count, not by bytes, so this caps what one
+/// poll can load: about 1 GB of the largest messages. A byte cap in Iggy would lift it.
+const MAX_PAGE_COUNT: u32 = 16;
+/// The largest message Iggy stores, header included.
+const LARGEST_MESSAGE_BYTES: u64 =
+ MAX_PAYLOAD_SIZE as u64 + MAX_USER_HEADERS_SIZE as u64 + IGGY_MESSAGE_HEADER_SIZE as u64;
+// The server answers an empty poll when the reply passes its u32 frame size. The consumer would
+// then stall.
+const _: () = assert!(MAX_PAGE_COUNT as u64 * LARGEST_MESSAGE_BYTES < u32::MAX as u64);
+
+/// First version with sessions and a top-level error code.
+const SESSION_MIN_VERSION: i16 = 7;
+/// The epochs of a full request. Any other epoch continues a session.
+const INITIAL_EPOCH: i32 = 0;
+const FINAL_EPOCH: i32 = -1;
+
+/// Without a bridge, the stub. With one, records from Iggy.
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
- handle_versioned_request(
+ let decoded = decode_request(
API_KEY_FETCH,
api_version,
body,
- |v, b| {
- decode_guarded::<FetchRequest>(v, b, |v, b| {
- validate_fetch_shape(v, b, state.max_frame_size)
- })
- },
- encode_response,
+ |v, b| decode(state, v, b),
encode_error_response,
"Fetch",
- )
+ );
+ let request = match decoded {
+ Ok(request) => request,
+ Err(outcome) => return outcome,
+ };
+ let Some(bridge) = &state.bridge else {
+ return respond_or_close(encode_response(api_version, &request), "Fetch");
+ };
+ // Kafka answers an unknown session the same way. The client then sends a full request.
+ if !is_full(api_version, &request) {
+ tracing::debug!(
+ session_id = request.session_id,
+ session_epoch = request.session_epoch,
+ "Fetch continues a session this gateway never opened"
+ );
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_FETCH_SESSION_ID_NOT_FOUND),
+ "Fetch",
+ );
+ }
+
+ let (responses, slot) = fetch(bridge, state, &request).await;
+ let record_bytes: usize = responses
+ .iter()
+ .flat_map(|topic| &topic.partitions)
+ .map(|partition| partition.records.as_ref().map_or(0, Bytes::len))
+ .sum();
+ // The records are encoded already, so this is a plain copy.
+ let encoded = off_worker(record_bytes >= COPY_OFF_WORKER_BYTES, || {
+ encode_inner(api_version, responses, ERROR_NONE)
+ });
+ // The slot covers the read and the encode. A slow socket must not hold it.
+ drop(slot);
+ respond_or_close(encoded, "Fetch")
}
/// Well-formed Fetch response. Uses top-level `error_code` at v7+, or a single
@@ -62,7 +143,7 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
///
/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
- if version >= 7 {
+ if version >= SESSION_MIN_VERSION {
return encode_inner(version, Vec::new(), error_code);
}
// No top-level error field below v7; the error surfaces on the placeholder partition instead.
@@ -97,6 +178,1087 @@ pub fn encode_response(version: i16, req: &FetchRequest) -> Result<Bytes> {
encode_inner(version, topics, ERROR_NONE)
}
+fn decode(state: &GatewayState, version: i16, body: Bytes) -> Result<FetchRequest> {
+ decode_guarded::<FetchRequest>(version, body, |v, b| {
+ validate_fetch_shape(v, b, state.max_frame_size)
+ })
+}
+
+/// Whether `request` names every partition it wants. Anything else continues a session, and
+/// this gateway opens none: it always answers `session_id` 0.
+const fn is_full(version: i16, request: &FetchRequest) -> bool {
+ version < SESSION_MIN_VERSION || matches!(request.session_epoch, INITIAL_EPOCH | FINAL_EPOCH)
+}
+
+/// Reads every requested partition once. When that finds nothing, waits for records up to
+/// `max_wait_ms` and reads again.
+///
+/// Returns the read slot too, if the read took one. Keep it until the response is encoded.
+async fn fetch(
+ bridge: &Arc<IggyBridge>,
+ state: &GatewayState,
+ request: &FetchRequest,
+) -> (Vec<FetchableTopicResponse>, Option<FetchSlot>) {
+ let started = Instant::now();
+ let give_up_at = started + REQUEST_DEADLINE;
+ let fetch = Fetch::new(bridge, state, request, give_up_at);
+ // A shared probe will do: an offset past it goes to a poll, not straight to a 1.
+ let mut probes = fetch.probe_all(recent(started)).await;
+ let (mut answers, mut slot) = fetch.read(&probes).await;
+ if waits(request, &answers) {
+ // No slot while it waits.
+ drop(slot);
+ let held: Vec<bool> = answers
+ .iter()
+ .map(|answer| retried_at_once(answer.error_code))
+ .collect();
+ let wait_until = started + fetch.max_wait;
+ probes = fetch
+ .wait(probes, &held, request.min_bytes, wait_until)
+ .await;
+ // Kafka reads again when the wait ends, too.
+ (answers, slot) = fetch.read(&probes).await;
+ }
+ (group(request, &fetch.wanted, answers), slot)
+}
+
+/// Kafka answers at once when a partition fails, when the request allows no wait, or when the
+/// records read reach `min_bytes`. Otherwise it waits.
+///
+/// A 6 or a -1 waits too. See [`retried_at_once`].
+fn waits(request: &FetchRequest, answers: &[PartitionData]) -> bool {
+ let min_bytes = usize::try_from(request.min_bytes).unwrap_or(0);
+ let record_bytes: usize = answers
+ .iter()
+ .map(|answer| answer.records.as_ref().map_or(0, Bytes::len))
+ .sum();
+ request.max_wait_ms > 0
+ && !answers.is_empty()
+ && record_bytes < min_bytes
+ && answers
+ .iter()
+ .all(|answer| answer.error_code == ERROR_NONE || retried_at_once(answer.error_code))
+}
+
+/// Codes the Java consumer fetches again at once after. Sent at once, they make it spin, so they
+/// wait out the request instead.
+const fn retried_at_once(code: i16) -> bool {
+ matches!(
+ code,
+ ERROR_NOT_LEADER_OR_FOLLOWER | ERROR_UNKNOWN_SERVER_ERROR
+ )
+}
+
+fn wait_time(max_wait_ms: i32) -> Duration {
+ Duration::from_millis(u64::try_from(max_wait_ms).unwrap_or(0)).min(MAX_WAIT)
+}
+
+/// When a -1 hold that starts at `now` ends: after the request's wait, but at least
+/// [`PROBE_INTERVAL`] on. A consumer with a short wait then reads a stuck partition at most ten
+/// times a second.
+fn hold_until(now: Instant, max_wait: Duration) -> Instant {
+ now + max_wait.max(PROBE_INTERVAL)
+}
+
+/// The topics to load a new probe for: each one whose newest probe started before `since`, the
+/// oldest first, at most [`MAX_ROUND_TOPICS`].
+fn stale_topics(snapshots: &[Option<Arc<Snapshot>>], since: Instant) -> Vec<usize> {
+ let started = |topic: usize| snapshots[topic].as_ref().map(|snapshot| snapshot.started);
+ let mut stale: Vec<usize> = (0..snapshots.len())
+ .filter(|&topic| started(topic).is_none_or(|started| started < since))
+ .collect();
+ // No probe sorts first, then the oldest, so each topic gets its turn.
+ stale.sort_by_key(|&topic| started(topic));
+ stale.truncate(MAX_ROUND_TOPICS);
+ stale
+}
+
+/// The topics wait round `turn` probes: the next [`MAX_ROUND_TOPICS`] of `count`, in turn.
+fn round(turn: usize, count: usize) -> impl Iterator<Item = usize> {
+ let start = turn.saturating_mul(MAX_ROUND_TOPICS) % count.max(1);
+ (start..start + MAX_ROUND_TOPICS.min(count)).map(move |topic| topic % count)
+}
+
+/// Waits for the next gateway write to any of the topics behind `writes`, then returns the topics
+/// written, at most [`MAX_ROUND_TOPICS`], each with the time of its write, and marks them seen.
+/// Pends while no write can come.
+async fn next_writes(writes: &mut [watch::Receiver<Option<Instant>>]) -> Vec<(usize, Instant)> {
+ let first = {
+ let mut changes: Vec<_> = writes
+ .iter_mut()
+ .map(|write| Some(Box::pin(write.changed())))
+ .collect();
+ poll_fn(|context| {
+ for (topic, slot) in changes.iter_mut().enumerate() {
+ let Some(change) = slot else { continue };
+ match change.as_mut().poll(context) {
+ // `changed` marks this write seen.
+ Poll::Ready(Ok(())) => return Poll::Ready(topic),
+ // The topic left the board, so no write wakes this one.
+ Poll::Ready(Err(_)) => *slot = None,
+ Poll::Pending => {}
+ }
+ }
+ Poll::Pending
+ })
+ .await
+ };
+ let mut written = vec![(first, written_at(&mut writes[first]))];
+ written.extend(
+ writes
+ .iter_mut()
+ .enumerate()
+ .filter(|(topic, write)| *topic != first && write.has_changed().unwrap_or(false))
+ .take(MAX_ROUND_TOPICS - 1)
+ .map(|(topic, write)| (topic, written_at(write))),
+ );
+ written
+}
+
+/// When the newest write that `write` shows ended. Marks it seen.
+fn written_at(write: &mut watch::Receiver<Option<Instant>>) -> Instant {
+ let seen = *write.borrow_and_update();
+ seen.unwrap_or_else(Instant::now)
+}
+
+/// One request: what it asks for, and what bounds it.
+struct Fetch<'a> {
+ bridge: &'a Arc<IggyBridge>,
+ state: &'a GatewayState,
+ wanted: Vec<Wanted<'a>>,
+ /// Each topic to probe once, in request order.
+ topics: Vec<&'a str>,
+ response_bytes: usize,
+ give_up_at: Instant,
+ /// The request's wait, at most [`MAX_WAIT`].
+ max_wait: Duration,
+}
+
+impl<'a> Fetch<'a> {
+ fn new(
+ bridge: &'a Arc<IggyBridge>,
+ state: &'a GatewayState,
+ request: &'a FetchRequest,
+ give_up_at: Instant,
+ ) -> Self {
+ let (wanted, topics) = wanted(request);
+ Self {
+ bridge,
+ state,
+ wanted,
+ topics,
+ response_bytes: response_bytes(request.max_bytes, state.max_frame_size),
+ give_up_at,
+ max_wait: wait_time(request.max_wait_ms),
+ }
+ }
+
+ /// Each topic's probe, started at `since` or later.
+ ///
+ /// Loads at most [`MAX_ROUND_TOPICS`] new probes, the oldest first. Other topics keep an older
+ /// probe, or answer 6 without one. So does a topic not probed before `give_up_at`.
+ async fn probe_all(&self, since: Instant) -> Probes {
+ let board = &self.state.probe_board;
+ let mut snapshots: Vec<_> = self
+ .topics
+ .iter()
+ .map(|&topic| board.newest(topic))
+ .collect();
+ for topic in stale_topics(&snapshots, since) {
+ snapshots[topic] = self
+ .probe_topic(self.topics[topic], since, self.give_up_at)
+ .await;
+ }
+ snapshots
+ .into_iter()
+ .map(|snapshot| {
+ snapshot.map_or(Err(ERROR_NOT_LEADER_OR_FOLLOWER), |snapshot| {
+ snapshot.probe.clone()
+ })
+ })
+ .collect()
+ }
+
+ /// `topic`'s probe from the board, started at `since` or later. `None` if none comes by
+ /// `deadline`.
+ async fn probe_topic(
+ &self,
+ topic: &str,
+ since: Instant,
+ deadline: Instant,
+ ) -> Option<Arc<Snapshot>> {
+ let load = || {
+ let (bridge, name) = (Arc::clone(self.bridge), topic.to_owned());
+ async move {
+ let probe = bridge.probe(&name).await;
+ probe.map(Arc::new).map_err(|error| refusal(&name, &error))
+ }
+ };
+ self.state
+ .probe_board
+ .probe(topic, since, deadline, load)
+ .await
+ }
+
+ /// The probe of partition `index` of `topic`, started at `since` or later.
+ async fn probe_partition(
+ &self,
+ topic: &str,
+ index: u32,
+ since: Instant,
+ ) -> Option<PartitionProbe> {
+ let snapshot = self.probe_topic(topic, since, self.give_up_at).await?;
+ snapshot.probe.as_ref().ok()?.get(index)
+ }
+
+ /// Probes until the bytes waiting reach `min_bytes`, a partition fails, or `wait_until`
+ /// passes. Holds no slot while it waits.
+ ///
+ /// Each [`PROBE_INTERVAL`], a round probes the next [`MAX_ROUND_TOPICS`] topics. A gateway
+ /// write to a topic probes it at once. A topic not probed keeps its last probe.
+ ///
+ /// `held` marks partitions that answered 6 or -1. They wait it out and end no wait.
+ async fn wait(
+ &self,
+ mut probes: Probes,
+ held: &[bool],
+ min_bytes: i32,
+ wait_until: Instant,
+ ) -> Probes {
+ if held.iter().all(|&held| held) {
+ sleep_until(wait_until).await;
+ return self.probe_all(recent(Instant::now())).await;
+ }
+ let min_bytes = u64::try_from(min_bytes).unwrap_or(0);
+ let board = &self.state.probe_board;
+ let mut writes: Vec<_> = self
+ .topics
+ .iter()
+ .map(|&topic| board.writes(topic))
+ .collect();
+ let mut turn = 0;
+ // Kept across write wakes, so steady writes to one topic do not stop the rounds.
+ let mut round_at = (Instant::now() + PROBE_INTERVAL).min(wait_until);
+ while Instant::now() < wait_until {
+ let written = tokio::select! {
+ () = sleep_until(round_at) => None,
+ written = next_writes(&mut writes) => Some(written),
+ };
+ let timed = written.is_none();
+ // A write needs a probe that started after it. A round needs one newer than the last.
+ let due = written.unwrap_or_else(|| {
+ turn += 1;
+ let since = recent(Instant::now());
+ round(turn - 1, self.topics.len())
+ .map(|topic| (topic, since))
+ .collect()
+ });
+ for (topic, since) in due {
+ // A probe late for the round still lands on the board for the next Fetch.
+ let probe = self.probe_topic(self.topics[topic], since, wait_until);
+ if let Some(snapshot) = probe.await {
+ probes[topic].clone_from(&snapshot.probe);
+ }
+ }
+ if timed {
+ round_at = (Instant::now() + PROBE_INTERVAL).min(wait_until);
+ }
+ if ready(&self.wanted, held, &probes, self.response_bytes, min_bytes) {
+ break;
+ }
+ }
+ probes
+ }
+
+ /// One answer per requested partition, in request order, and the read slot, if taken.
+ async fn read(&self, probes: &Probes) -> (Vec<PartitionData>, Option<FetchSlot>) {
+ Reader::new(self).read_all(probes).await
+ }
+}
+
+/// One requested partition.
+struct Wanted<'a> {
+ /// Its topic entry in the request.
+ entry: usize,
+ topic: &'a str,
+ partition: i32,
+ fetch_offset: i64,
+ max_bytes: usize,
+ /// Where its probe is, or the code it answers without asking Iggy.
+ probe_at: core::result::Result<ProbeAt, i16>,
+}
+
+/// Where a partition's probe is.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ProbeAt {
+ /// Its topic's place in `Fetch::topics`.
+ topic: usize,
+ index: u32,
+}
+
+/// Each requested partition once, in request order, and each topic to probe once. A repeat
+/// keeps the first entry.
+///
+/// The Java consumer drops a whole response that leaves out a partition it asked for.
+fn wanted(request: &FetchRequest) -> (Vec<Wanted<'_>>, Vec<&str>) {
+ let total = request
+ .topics
+ .iter()
+ .map(|topic| topic.partitions.len())
+ .sum();
+ let mut seen = HashSet::with_capacity(total);
+ let mut wanted = Vec::with_capacity(total);
+ let mut topics = IndexSet::new();
+ for (entry, topic) in request.topics.iter().enumerate() {
+ let name = topic.topic.as_str();
+ // A name Kafka refuses answers 3 here, so it never takes a place on the probe board.
+ let named = validate_kafka_topic_name("kafka_topic", name)
+ .map_err(|_| ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ for partition in &topic.partitions {
+ if !seen.insert((name, partition.partition)) {
+ continue;
+ }
+ let probe_at = named
+ .and_then(|()| {
+ u32::try_from(partition.partition).map_err(|_| ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ })
+ .map(|index| ProbeAt {
+ topic: topics.insert_full(name).0,
+ index,
+ });
+ wanted.push(Wanted {
+ entry,
+ topic: name,
+ partition: partition.partition,
+ fetch_offset: partition.fetch_offset,
+ max_bytes: usize::try_from(partition.partition_max_bytes).unwrap_or(0),
+ probe_at,
+ });
+ }
+ }
+ (wanted, topics.into_iter().collect())
+}
+
+/// Record bytes the whole response may carry: `max_bytes`, at most one frame.
+fn response_bytes(max_bytes: i32, max_frame_size: usize) -> usize {
+ usize::try_from(max_bytes).unwrap_or(0).min(max_frame_size)
+}
+
+/// Per topic in `Fetch::topics`, its probe, or one code for the whole topic.
+type Probes = Vec<TopicProbed>;
+
+/// Whether a partition now fails, as an error ends Kafka's wait too, or the bytes waiting reach
+/// `min_bytes`. Skips `held` partitions.
+fn ready(
+ wanted: &[Wanted<'_>],
+ held: &[bool],
+ probes: &Probes,
+ response_bytes: usize,
+ min_bytes: u64,
+) -> bool {
+ let mut waiting = 0u64;
+ for (want, _) in wanted.iter().zip(held).filter(|&(_, &held)| !held) {
+ match position(want, probes) {
+ Position::Refused(_) => return true,
+ // Past the end of this probe, or no message kept. A newer probe, or the read after
+ // the wait, decides.
+ Position::CaughtUp(_) | Position::Unsure { .. } => {}
+ Position::Behind { offset, probe, .. } => {
+ let budget = want.max_bytes.min(response_bytes);
+ waiting = waiting.saturating_add(bytes_waiting(offset, probe, budget));
+ }
+ }
+ }
+ waiting >= min_bytes
+}
+
+/// Estimated bytes past `offset`, at most `budget`. Kafka's delayed fetch counts the same way.
+fn bytes_waiting(offset: u64, probe: PartitionProbe, budget: usize) -> u64 {
+ probe
+ .high_watermark
+ .saturating_sub(offset)
+ .saturating_mul(probe.average_size)
+ .min(u64::try_from(budget).unwrap_or(u64::MAX))
+}
+
+/// Where a requested offset stands against the probe.
+#[derive(Debug, PartialEq, Eq)]
+enum Position {
+ /// Answer this code, with no records.
+ Refused(i16),
+ /// At the high watermark. Answer 0, with no records.
+ CaughtUp(PartitionProbe),
+ /// Records wait in partition `index` from `offset` on.
+ Behind {
+ index: u32,
+ offset: u64,
+ probe: PartitionProbe,
+ },
+ /// The probe cannot place `offset`. One poll decides.
+ Unsure {
+ index: u32,
+ offset: u64,
+ probe: PartitionProbe,
+ },
+}
+
+fn position(want: &Wanted<'_>, probes: &Probes) -> Position {
+ let at = match want.probe_at {
+ Ok(at) => at,
+ Err(code) => return Position::Refused(code),
+ };
+ // `Fetch::probe_all` probes every topic in `Fetch::topics`, in order.
+ let probe = match &probes[at.topic] {
+ Ok(topic) => match topic.get(at.index) {
+ Some(probe) => probe,
+ None => return Position::Refused(ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ },
+ Err(code) => return Position::Refused(*code),
+ };
+ if probe.is_loading() {
+ return Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER);
+ }
+ // Below the oldest retained offset is in range: Iggy reads from the first retained message,
+ // and Kafka consumers accept the gap.
+ match u64::try_from(want.fetch_offset) {
+ Ok(offset) if offset == probe.high_watermark => Position::CaughtUp(probe),
+ // Retention can remove every message, and a server with no stats reads the same. Past
+ // the end, a shared probe can predate records the consumer already read.
+ Ok(offset) if probe.messages_count == 0 || offset > probe.high_watermark => {
+ Position::Unsure {
+ index: at.index,
+ offset,
+ probe,
+ }
+ }
+ Ok(offset) => Position::Behind {
+ index: at.index,
+ offset,
+ probe,
+ },
+ Err(_) => Position::Refused(ERROR_OFFSET_OUT_OF_RANGE),
+ }
+}
+
+/// Reads partitions in request order, as Kafka does from v3, and spends one byte budget on them.
+struct Reader<'a> {
+ fetch: &'a Fetch<'a>,
+ bytes_left: usize,
+ /// No partition has answered with records yet, so the next one returns at least one, whatever
+ /// the limits say (KIP-74). A record larger than every limit would stall the consumer
+ /// otherwise.
+ owes_first_record: bool,
+ /// Taken at the first poll. A poll given up on keeps it until the poll ends.
+ slot: Option<FetchSlot>,
+}
+
+impl<'a> Reader<'a> {
+ const fn new(fetch: &'a Fetch<'a>) -> Self {
+ Self {
+ fetch,
+ bytes_left: fetch.response_bytes,
+ owes_first_record: true,
+ slot: None,
+ }
+ }
+
+ /// One answer per requested partition, in request order, and the slot, if taken.
+ async fn read_all(mut self, probes: &Probes) -> (Vec<PartitionData>, Option<FetchSlot>) {
+ let fetch = self.fetch;
+ let mut answers = Vec::with_capacity(fetch.wanted.len());
+ for want in &fetch.wanted {
+ let answer = match position(want, probes) {
+ Position::Refused(code) => refused(want.partition, code),
+ Position::CaughtUp(probe) => {
+ served(want.partition, probe.high_watermark, Bytes::new())
+ }
+ Position::Behind { index, offset, .. } | Position::Unsure { index, offset, .. }
+ if held(
+ &fetch.state.stuck_offsets,
+ want.topic,
+ index,
+ offset,
+ Instant::now(),
+ ) =>
+ {
+ refused(want.partition, ERROR_UNKNOWN_SERVER_ERROR)
+ }
+ Position::Behind {
+ index,
+ offset,
+ probe,
+ } => {
+ let polled = Polled::new(offset, probe.high_watermark);
+ self.read(want, index, offset, probe, polled).await
+ }
+ Position::Unsure {
+ index,
+ offset,
+ probe,
+ } => self.settle(want, index, offset, probe).await,
+ };
+ answers.push(answer);
+ }
+ (answers, self.slot)
+ }
+
+ /// Polls one partition that has records waiting, and answers with what its budget holds.
+ async fn read(
+ &mut self,
+ want: &Wanted<'_>,
+ index: u32,
+ offset: u64,
+ probe: PartitionProbe,
+ mut polled: Polled,
+ ) -> PartitionData {
+ let budget = want.max_bytes.min(self.bytes_left);
+ // Unless a record is owed, skip a poll that likely brings back nothing that fits.
+ let average = usize::try_from(probe.average_size).unwrap_or(usize::MAX);
+ if self.owes_first_record || budget >= average.max(1) {
+ while let Some(count) = next_page(&polled, budget, probe.average_size) {
+ let asked = Instant::now();
+ let Some(page) = self.poll(want.topic, index, polled.next, count).await else {
+ break;
+ };
+ match page {
+ // The probe showed records, yet the first poll found none.
+ Ok(page) if page.messages.is_empty() && polled.messages.is_empty() => {
+ return self
+ .after_empty_poll(want, index, offset, probe, asked)
+ .await;
+ }
+ Ok(page) => polled.add(page, count),
+ Err(error) if polled.messages.is_empty() => {
+ return self.refuse(want, index, offset, &error);
+ }
+ // Serve what the earlier polls brought.
+ Err(_) => break,
+ }
+ }
+ }
+
+ let keep_one = self.owes_first_record;
+ let encoded = off_worker(polled.bytes >= CODEC_OFF_WORKER_BYTES, || {
+ encode_records(&polled.messages, offset, budget, keep_one)
+ });
+ match encoded {
+ Ok(batch) => {
+ self.bytes_left = self.bytes_left.saturating_sub(batch.len());
+ self.owes_first_record &= batch.is_empty();
+ served(want.partition, polled.high_watermark, batch)
+ }
+ Err(error) => {
+ let until = hold_until(Instant::now(), self.fetch.max_wait);
+ let stuck = &self.fetch.state.stuck_offsets;
+ let first = hold(stuck, want.topic, index, offset, until);
+ log_stuck(first, want.topic, index, offset, &error);
+ refused(want.partition, ERROR_UNKNOWN_SERVER_ERROR)
+ }
+ }
+ }
+
+ /// Reads a partition the probe cannot place, with a poll at `offset`.
+ ///
+ /// - Records: serve them. The probe missed them.
+ /// - Nothing, and Iggy's commit ends just before `offset`: caught up.
+ /// - Nothing else: a probe taken after the poll decides. See [`empty_poll`].
+ async fn settle(
+ &mut self,
+ want: &Wanted<'_>,
+ index: u32,
+ offset: u64,
+ probe: PartitionProbe,
+ ) -> PartitionData {
+ // Sized as a read's first page, since an empty answer costs the same round trip.
+ let budget = want.max_bytes.min(self.bytes_left);
+ let count = poll_count(budget, probe.average_size, u64::from(MAX_PAGE_COUNT));
+ let asked = Instant::now();
+ match self.poll(want.topic, index, offset, count).await {
+ // Never below the offset, or the consumer reports negative lag.
+ None => served(
+ want.partition,
+ probe.high_watermark.max(offset),
+ Bytes::new(),
+ ),
+ Some(Err(error)) => self.refuse(want, index, offset, &error),
+ Some(Ok(page)) if !page.messages.is_empty() => {
+ let mut polled = Polled::new(offset, probe.high_watermark);
+ polled.add(page, count);
+ self.read(want, index, offset, probe, polled).await
+ }
+ Some(Ok(page)) if caught_up(offset, page.current_offset) => {
+ served(want.partition, offset, Bytes::new())
+ }
+ Some(Ok(_)) => {
+ self.after_empty_poll(want, index, offset, probe, asked)
+ .await
+ }
+ }
+ }
+
+ /// Answers an empty poll at `offset`, made at `asked`, from `before` and a probe taken after
+ /// it. See [`empty_poll`].
+ async fn after_empty_poll(
+ &self,
+ want: &Wanted<'_>,
+ index: u32,
+ offset: u64,
+ before: PartitionProbe,
+ asked: Instant,
+ ) -> PartitionData {
+ let after = self.fetch.probe_partition(want.topic, index, asked).await;
+ match empty_poll(offset, before, after) {
+ EmptyPoll::OutOfRange { high_watermark } => {
+ // No offset past 0: maybe a partition that Iggy loads or fences.
+ let not_ready = &self.fetch.state.not_ready;
+ let loading = high_watermark <= 1
+ && spell(not_ready, want.topic, index, Instant::now()) < RELOAD_GRACE;
+ let code = if loading {
+ ERROR_NOT_LEADER_OR_FOLLOWER
+ } else {
+ ERROR_OFFSET_OUT_OF_RANGE
+ };
+ refused(want.partition, code)
+ }
+ EmptyPoll::NotReady => {
+ let spell = spell(
+ &self.fetch.state.not_ready,
+ want.topic,
+ index,
+ Instant::now(),
+ );
+ log_not_ready(spell.is_zero(), want.topic, index, offset);
+ refused(want.partition, ERROR_NOT_LEADER_OR_FOLLOWER)
+ }
+ EmptyPoll::Unproven => {
+ let high_watermark =
+ after.map_or(before.high_watermark, |after| after.high_watermark);
+ served(want.partition, high_watermark.max(offset), Bytes::new())
+ }
+ }
+ }
+
+ /// The code for `error`. A -1 also holds the partition, so reads skip Iggy until the hold ends.
+ fn refuse(
+ &self,
+ want: &Wanted<'_>,
+ index: u32,
+ offset: u64,
+ error: &BridgeError,
+ ) -> PartitionData {
+ let code = refusal(want.topic, error);
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ let until = hold_until(Instant::now(), self.fetch.max_wait);
+ hold(
+ &self.fetch.state.stuck_offsets,
+ want.topic,
+ index,
+ offset,
+ until,
+ );
+ }
+ refused(want.partition, code)
+ }
+
+ /// Polls one page, in the read slot. `None` once `give_up_at` passes.
+ async fn poll(
+ &mut self,
+ topic: &str,
+ index: u32,
+ offset: u64,
+ count: u32,
+ ) -> Option<core::result::Result<PolledMessages, BridgeError>> {
+ let give_up_at = self.fetch.give_up_at;
+ if Instant::now() >= give_up_at {
+ return None;
+ }
+ let bridge = self.fetch.bridge;
+ let (page, slot) = bridge
+ .poll(
+ self.take_slot().await?,
+ topic,
+ index,
+ offset,
+ count,
+ give_up_at,
+ )
+ .await?;
+ self.slot = Some(slot);
+ Some(page)
+ }
+
+ /// The read slot, taken at the first poll. `None` once `give_up_at` passes.
+ async fn take_slot(&mut self) -> Option<FetchSlot> {
+ match self.slot.take() {
+ Some(slot) => Some(slot),
+ None => self.fetch.bridge.fetch_slot(self.fetch.give_up_at).await,
+ }
+ }
+}
+
+/// Whether an empty poll at `offset` proves the consumer is at the end: Iggy's commit counter,
+/// `current_offset`, ends just before it. A 0 proves nothing, since an empty partition and a failed
+/// read send 0 too.
+const fn caught_up(offset: u64, current_offset: u64) -> bool {
+ current_offset > 0 && current_offset.saturating_add(1) == offset
+}
+
+/// What an empty poll proves.
+#[derive(Debug, PartialEq, Eq)]
+enum EmptyPoll {
+ /// The offset is past `high_watermark`, the end after the poll.
+ OutOfRange { high_watermark: u64 },
+ /// Records remain, yet the poll found none: a read fault, or a partition Iggy loads. Retry.
+ NotReady,
+ /// Nothing. Serve no records, and the consumer asks again.
+ Unproven,
+}
+
+/// Decides an empty poll at `offset` from the probe before it and one taken after it.
+///
+/// Iggy also answers empty on a read fault, so only `after` proves anything. No message kept below
+/// the end proves no gap: a cluster replica reads so after a crash or a failed install, while its
+/// peers keep the records. So the consumer waits for the next write instead of moving past them.
+const fn empty_poll(
+ offset: u64,
+ before: PartitionProbe,
+ after: Option<PartitionProbe>,
+) -> EmptyPoll {
+ let Some(after) = after else {
+ return EmptyPoll::Unproven;
+ };
+ if after.is_loading() {
+ EmptyPoll::NotReady
+ } else if offset > after.high_watermark {
+ EmptyPoll::OutOfRange {
+ high_watermark: after.high_watermark,
+ }
+ } else if after.messages_count > 0
+ && offset < before.high_watermark
+ && offset < after.high_watermark
+ {
+ EmptyPoll::NotReady
+ } else {
+ EmptyPoll::Unproven
+ }
+}
+
+/// Per Kafka topic and partition, a spell of answers that found it not ready.
+#[derive(Debug, Default)]
+pub(crate) struct Spells {
+ by_partition: HashMap<(String, u32), Spell>,
+ /// The size at which the next new spell sweeps first.
+ sweep_at: usize,
+}
+
+/// No sweep below this many spells.
+const MIN_SPELL_SWEEP: usize = 1024;
+
+#[derive(Debug, Clone, Copy)]
+struct Spell {
+ first: Instant,
+ last: Instant,
+}
+
+/// How long `partition` of `topic` reads as not ready, as of `now`. Zero for a new spell. A spell
+/// ends once the partition goes [`RELOAD_GRACE`] without reading so.
+fn spell(spells: &Mutex<Spells>, topic: &str, partition: u32, now: Instant) -> Duration {
+ let key = (topic.to_owned(), partition);
+ let mut spells = spells.lock().unwrap_or_else(PoisonError::into_inner);
+ if !spells.by_partition.contains_key(&key) && spells.by_partition.len() >= spells.sweep_at {
+ spells
+ .by_partition
+ .retain(|_, spell| now.saturating_duration_since(spell.last) < RELOAD_GRACE);
+ // Twice what is left, so sweeps cost O(1) per new spell.
+ spells.sweep_at = spells
+ .by_partition
+ .len()
+ .saturating_mul(2)
+ .max(MIN_SPELL_SWEEP);
+ }
+ let spell = spells.by_partition.entry(key).or_insert(Spell {
+ first: now,
+ last: now,
+ });
+ if now.saturating_duration_since(spell.last) >= RELOAD_GRACE {
+ spell.first = now;
+ }
+ spell.last = now;
+ let first = spell.first;
+ drop(spells);
+ now.saturating_duration_since(first)
+}
+
+/// Logs a partition whose records Iggy did not return: `warn!` when a spell starts, `debug!` while
+/// it lasts.
+fn log_not_ready(first: bool, topic: &str, partition: u32, offset: u64) {
+ if first {
+ tracing::warn!(
+ kafka_topic = topic,
+ partition,
+ fetch_offset = offset,
+ "Fetch found no records where Iggy reports some"
+ );
+ } else {
+ tracing::debug!(
+ kafka_topic = topic,
+ partition,
+ fetch_offset = offset,
+ "Fetch still finds no records where Iggy reports some"
+ );
+ }
+}
+
+/// Per Kafka topic, then partition, where a partition last answered -1.
+pub(crate) type StuckOffsets = Mutex<HashMap<String, HashMap<u32, Stuck>>>;
+
+/// A partition answered -1 at `offset`. Until `until`, it answers -1 there without a read.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Stuck {
+ offset: u64,
+ until: Instant,
+}
+
+/// Holds `partition` of `topic` at `offset` until `until`. `true` if it did not stop there before.
+fn hold(stuck: &StuckOffsets, topic: &str, partition: u32, offset: u64, until: Instant) -> bool {
+ let before = stuck
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .entry(topic.to_owned())
+ .or_default()
+ .insert(partition, Stuck { offset, until });
+ before.is_none_or(|before| before.offset != offset)
+}
+
+/// Whether `partition` of `topic` answered -1 at `offset` and its hold runs at `now`.
+fn held(stuck: &StuckOffsets, topic: &str, partition: u32, offset: u64, now: Instant) -> bool {
+ stuck
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .get(topic)
+ .and_then(|partitions| partitions.get(&partition))
+ .is_some_and(|held| held.offset == offset && now < held.until)
+}
+
+/// Logs a stored message that does not map: `warn!` the first time a partition stops at an offset,
+/// `debug!` while it stays there. The consumer asks for that offset again until someone acts.
+fn log_stuck(first: bool, topic: &str, partition: u32, offset: u64, error: &RecordCodecError) {
+ if first {
+ tracing::warn!(
+ %error,
+ kafka_topic = topic,
+ partition,
+ fetch_offset = offset,
+ "Fetch cannot map a stored message"
+ );
+ } else {
+ tracing::debug!(
+ %error,
+ kafka_topic = topic,
+ partition,
+ fetch_offset = offset,
+ "Fetch still cannot map a stored message"
+ );
+ }
+}
+
+/// What one partition read polled so far.
+struct Polled {
+ messages: Vec<IggyMessage>,
+ /// Where the next page starts.
+ next: u64,
+ /// Stored bytes of `messages`, and of the largest one.
+ bytes: usize,
+ largest: usize,
+ high_watermark: u64,
+ /// A page came back short, so nothing more waits.
+ done: bool,
+}
+
+impl Polled {
+ const fn new(offset: u64, high_watermark: u64) -> Self {
+ Self {
+ messages: Vec::new(),
+ next: offset,
+ bytes: 0,
+ largest: 0,
+ high_watermark,
+ done: false,
+ }
+ }
+
+ fn add(&mut self, page: PolledMessages, count: u32) {
+ self.done = page.messages.len() < usize::try_from(count).unwrap_or(usize::MAX);
+ let Some(last) = page.messages.last() else {
+ return;
+ };
+ self.next = last.header.offset.saturating_add(1);
+ self.high_watermark = self
+ .high_watermark
+ .max(page.current_offset.saturating_add(1));
+ for message in &page.messages {
+ let size = stored_bytes(message);
+ self.bytes = self.bytes.saturating_add(size);
+ self.largest = self.largest.max(size);
+ }
+ self.messages.extend(page.messages);
+ }
+}
+
+/// Messages to ask for in the next poll of one partition read, or `None` when it is done.
+///
+/// The first poll trusts the partition's average message size, the later ones the largest
+/// message seen so far. Large messages after small ones then cost one short poll at most.
+fn next_page(polled: &Polled, budget: usize, average_size: u64) -> Option<u32> {
+ let waiting = polled.high_watermark.saturating_sub(polled.next);
+ let taken = u32::try_from(polled.messages.len()).unwrap_or(u32::MAX);
+ let room = MAX_POLL_COUNT
+ .saturating_sub(taken)
+ .min(u32::try_from(waiting).unwrap_or(u32::MAX));
+ if polled.done || room == 0 {
+ return None;
+ }
+ if polled.messages.is_empty() {
+ return Some(poll_count(budget, average_size, waiting).min(MAX_PAGE_COUNT));
+ }
+ let fits = budget.saturating_sub(polled.bytes) / polled.largest.max(1);
+ let count = u32::try_from(fits)
+ .unwrap_or(u32::MAX)
+ .min(MAX_PAGE_COUNT)
+ .min(room);
+ (count > 0).then_some(count)
+}
+
+/// Messages to poll for about `budget` bytes at `average_size` bytes each.
+///
+/// At least one, since a record past the budget may be owed. At most what waits, and
+/// [`MAX_POLL_COUNT`].
+fn poll_count(budget: usize, average_size: u64, waiting: u64) -> u32 {
+ let fits = u64::try_from(budget).unwrap_or(u64::MAX) / average_size.max(1);
+ let count = fits.min(waiting).clamp(1, u64::from(MAX_POLL_COUNT));
+ u32::try_from(count).unwrap_or(MAX_POLL_COUNT)
+}
+
+/// Bytes Iggy stores for `message`, header included, as the probe's average counts them.
+fn stored_bytes(message: &IggyMessage) -> usize {
+ IGGY_MESSAGE_HEADER_SIZE
+ + message.payload.len()
+ + message.user_headers.as_ref().map_or(0, Bytes::len)
+}
+
+/// `messages` from `offset` on, as one batch of at most `budget` bytes.
+///
+/// `keep_one` keeps the first record even past `budget`. A message that does not map ends the
+/// batch before it, and fails it when it comes first.
+fn encode_records(
+ messages: &[IggyMessage],
+ offset: u64,
+ budget: usize,
+ keep_one: bool,
+) -> core::result::Result<Bytes, RecordCodecError> {
+ let mut records: Vec<Record> = Vec::with_capacity(messages.len());
+ let mut size = BATCH_HEADER_BYTES;
+ for message in messages
+ .iter()
+ .filter(|message| message.header.offset >= offset)
+ {
+ let Ok(kafka_offset) = i64::try_from(message.header.offset) else {
+ break;
+ };
+ let record = match from_iggy(message, kafka_offset) {
+ Ok(record) => record,
+ Err(error) if records.is_empty() => return Err(error),
+ // The next Fetch starts at this message and reports it.
+ Err(_) => break,
+ };
+ size = size.saturating_add(record_size_bound(&record));
+ if size > budget && !(keep_one && records.is_empty()) {
+ break;
+ }
+ records.push(record);
+ }
+ encode_batch(&mut records)
+}
+
+/// The Fetch code for a bridge error.
+///
+/// The Java consumer retries 3, 6 and -1, and throws on 29 on purpose. It throws on other codes
+/// too, so the rest fold into these.
+const fn fetch_code(error: &BridgeError) -> i16 {
+ match error.to_kafka_error_code() {
+ code @ (ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ | ERROR_NOT_LEADER_OR_FOLLOWER
+ | ERROR_TOPIC_AUTHORIZATION_FAILED) => code,
+ // A read changes nothing, so an unknown outcome is a plain retry.
+ ERROR_REQUEST_TIMED_OUT => ERROR_NOT_LEADER_OR_FOLLOWER,
+ // No topic has a name Kafka refuses.
+ ERROR_INVALID_TOPIC_EXCEPTION => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ _ => ERROR_UNKNOWN_SERVER_ERROR,
+ }
+}
+
+/// [`fetch_code`], logged at a level that matches who can fix it.
+fn refusal(kafka_topic: &str, error: &BridgeError) -> i16 {
+ let code = fetch_code(error);
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ // The client sees only -1, so this log is the one place the cause shows.
+ tracing::error!(%error, kafka_topic, "Iggy refused a Fetch read");
+ } else {
+ // The code names the cause: the client retries a 3 or a 6, and stops on a 29.
+ tracing::debug!(%error, kafka_topic, "Iggy refused a Fetch read");
+ }
+ code
+}
+
+/// Answers under their request topic entries. An entry left with no partition is dropped.
+fn group(
+ request: &FetchRequest,
+ wanted: &[Wanted<'_>],
+ answers: Vec<PartitionData>,
+) -> Vec<FetchableTopicResponse> {
+ let mut responses: Vec<FetchableTopicResponse> = Vec::new();
+ let mut open = None;
+ for (want, answer) in wanted.iter().zip(answers) {
+ if open != Some(want.entry) {
+ open = Some(want.entry);
+ responses.push(
+ FetchableTopicResponse::default()
+ .with_topic(request.topics[want.entry].topic.clone()),
+ );
+ }
+ if let Some(response) = responses.last_mut() {
+ response.partitions.push(answer);
+ }
+ }
+ responses
+}
+
+/// A partition answered with `code`: no records, and no offset worth trusting.
+fn refused(partition: i32, code: i16) -> PartitionData {
+ PartitionData::default()
+ .with_partition_index(partition)
+ .with_error_code(code)
+ .with_high_watermark(UNKNOWN_OFFSET)
+ .with_last_stable_offset(UNKNOWN_OFFSET)
+ .with_log_start_offset(UNKNOWN_OFFSET)
+ .with_aborted_transactions(None)
+ .with_records(Some(Bytes::new()))
+}
+
+/// A partition answered with `records`, which may be empty.
+///
+/// No transactions, so the last stable offset is the high watermark. The log start stays unknown.
+fn served(partition: i32, high_watermark: u64, records: Bytes) -> PartitionData {
+ let high_watermark = i64::try_from(high_watermark).unwrap_or(UNKNOWN_OFFSET);
+ PartitionData::default()
+ .with_partition_index(partition)
+ .with_high_watermark(high_watermark)
+ .with_last_stable_offset(high_watermark)
+ .with_log_start_offset(UNKNOWN_OFFSET)
+ .with_aborted_transactions(None)
+ .with_records(Some(records))
+}
+
fn encode_inner(
version: i16,
topics: Vec<FetchableTopicResponse>,
@@ -105,7 +1267,9 @@ fn encode_inner(
let resp = FetchResponse::default()
.with_error_code(top_level_error)
.with_responses(topics);
- encode_message(&resp, version, 512)
+ // Only a size hint. If the size fails, the encode fails and says why.
+ let capacity = resp.compute_size(version).unwrap_or(0);
+ encode_message(&resp, version, capacity)
}
fn partition_response(partition: i32, error_code: i16) -> PartitionData {
@@ -116,3 +1280,825 @@ fn partition_response(partition: i32, error_code: i16) -> PartitionData {
.with_log_start_offset(0)
.with_records(None)
}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::BTreeMap;
+
+ use iggy::prelude::{HeaderKey, HeaderValue, IggyError};
+ use kafka_protocol::indexmap::IndexMap;
+ use kafka_protocol::messages::TopicName;
+ use kafka_protocol::messages::fetch_request::{FetchPartition, FetchTopic};
+ use kafka_protocol::protocol::StrBytes;
+ use kafka_protocol::records::{
+ NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID, NO_SEQUENCE,
+ RecordBatchDecoder, TimestampType,
+ };
+
+ use super::*;
+ use crate::bridge::TopicProbe;
+ use crate::protocol::probe_board::ProbeBoard;
+ use crate::records::{MAPPING_VERSION, TimestampWindow, VERSION_HEADER, to_iggy};
+
+ const CREATE_TIME: i64 = 1_700_000_000_123;
+ /// Every partition code Fetch sends. The Java consumer throws on 29 on purpose, and on 1
+ /// when it has no reset policy.
+ const CONSUMER_CODES: [i16; 6] = [0, 1, 3, 6, 29, -1];
+
+ const fn probe(high_watermark: u64, average_size: u64) -> PartitionProbe {
+ PartitionProbe {
+ high_watermark,
+ average_size,
+ messages_count: if average_size == 0 { 0 } else { high_watermark },
+ }
+ }
+
+ /// A message at `offset` holding `value`, as Produce stores it.
+ fn stored(offset: u64, value: &[u8]) -> IggyMessage {
+ let 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: 0,
+ sequence: NO_SEQUENCE,
+ timestamp: CREATE_TIME,
+ key: None,
+ value: Some(Bytes::copy_from_slice(value)),
+ headers: IndexMap::new(),
+ };
+ let window = TimestampWindow::of(std::slice::from_ref(&record));
+ let mut message = to_iggy(&record, window).unwrap();
+ message.header.offset = offset;
+ message
+ }
+
+ /// A message that claims a mapping version this build does not read.
+ fn unmappable(offset: u64) -> IggyMessage {
+ let mut headers = BTreeMap::new();
+ headers.insert(
+ HeaderKey::try_from(VERSION_HEADER).unwrap(),
+ HeaderValue::try_from(&[MAPPING_VERSION + 1][..]).unwrap(),
+ );
+ let mut message = IggyMessage::builder()
+ .payload(Bytes::from_static(b"v"))
+ .user_headers(headers)
+ .build()
+ .unwrap();
+ message.header.offset = offset;
+ message
+ }
+
+ fn offsets(mut batch: Bytes) -> Vec<i64> {
+ RecordBatchDecoder::decode_all(&mut batch)
+ .unwrap()
+ .into_iter()
+ .flat_map(|set| set.records)
+ .map(|record| record.offset)
+ .collect()
+ }
+
+ /// Partition `partition` of the first topic probed.
+ fn want(partition: i32, fetch_offset: i64) -> Wanted<'static> {
+ Wanted {
+ entry: 0,
+ topic: "a",
+ partition,
+ fetch_offset,
+ max_bytes: 1024 * 1024,
+ probe_at: u32::try_from(partition)
+ .map(|index| ProbeAt { topic: 0, index })
+ .map_err(|_| ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ }
+ }
+
+ /// One topic, probed with `found`.
+ fn probes(found: &[(u32, PartitionProbe)]) -> Probes {
+ vec![Ok(Arc::new(found.iter().copied().collect()))]
+ }
+
+ fn request(topics: &[(&'static str, &[i32])]) -> FetchRequest {
+ FetchRequest::default().with_topics(
+ topics
+ .iter()
+ .map(|(name, partitions)| {
+ FetchTopic::default()
+ .with_topic(TopicName(StrBytes::from_static_str(name)))
+ .with_partitions(
+ partitions
+ .iter()
+ .map(|&partition| {
+ FetchPartition::default().with_partition(partition)
+ })
+ .collect(),
+ )
+ })
+ .collect(),
+ )
+ }
+
+ #[test]
+ fn given_a_budget_when_counting_should_poll_what_fits_at_the_average_size() {
+ assert_eq!(poll_count(1000, 100, 50), 10);
+ assert_eq!(poll_count(1000, 100, 3), 3, "never more than waits");
+ assert_eq!(
+ poll_count(usize::MAX, 1, u64::MAX),
+ MAX_POLL_COUNT,
+ "never more than the cap"
+ );
+ assert_eq!(poll_count(10, 100, 50), 1, "one, in case it is owed");
+ assert_eq!(poll_count(0, 100, 50), 1);
+ assert_eq!(
+ poll_count(500, 0, 50),
+ 50,
+ "no retained message to measure reads as one byte each"
+ );
+ }
+
+ /// A page of messages from `first` on, one per payload size.
+ fn page(first: u64, sizes: &[usize], current_offset: u64) -> PolledMessages {
+ let messages: Vec<IggyMessage> = sizes
+ .iter()
+ .zip(first..)
+ .map(|(&size, offset)| {
+ let mut message = IggyMessage::builder()
+ .payload(Bytes::from(vec![b'v'; size]))
+ .build()
+ .unwrap();
+ message.header.offset = offset;
+ message
+ })
+ .collect();
+ PolledMessages {
+ partition_id: 0,
+ current_offset,
+ count: u32::try_from(messages.len()).unwrap(),
+ messages,
+ }
+ }
+
+ #[test]
+ fn given_a_new_read_when_sizing_the_first_page_should_use_the_average_up_to_the_cap() {
+ let mib = 1024 * 1024;
+ assert_eq!(next_page(&Polled::new(0, 10_000), mib, 100), Some(16));
+ assert_eq!(next_page(&Polled::new(0, 3), mib, 100), Some(3));
+ assert_eq!(
+ next_page(&Polled::new(0, 100), 10, 100),
+ Some(1),
+ "one, in case it is owed"
+ );
+ assert_eq!(next_page(&Polled::new(5, 5), mib, 100), None, "none waits");
+ }
+
+ #[test]
+ fn given_small_messages_when_sizing_later_pages_should_grow_to_the_cap() {
+ let mut polled = Polled::new(0, 10_000);
+ polled.add(page(0, &[100; 16], 15), 16);
+ assert_eq!(polled.next, 16);
+ assert_eq!(next_page(&polled, 1024 * 1024, 100), Some(MAX_PAGE_COUNT));
+ }
+
+ #[test]
+ fn given_a_large_message_when_sizing_the_next_page_should_ask_for_what_fits() {
+ let mut polled = Polled::new(0, 10_000);
+ polled.add(page(0, &[100, 100, 400_000], 2), 3);
+ assert_eq!(next_page(&polled, 1024 * 1024, 100), Some(1));
+ assert_eq!(next_page(&polled, 500_000, 100), None, "no room left");
+ }
+
+ #[test]
+ fn given_a_short_page_when_sizing_the_next_page_should_stop() {
+ let mut polled = Polled::new(0, 10_000);
+ polled.add(page(0, &[100; 3], 20_000), 16);
+ assert_eq!(polled.high_watermark, 20_001, "the reply moves it up");
+ assert_eq!(next_page(&polled, 1024 * 1024, 100), None);
+ }
+
+ #[test]
+ fn given_the_count_cap_when_sizing_the_next_page_should_not_pass_it() {
+ let mut polled = Polled::new(0, 10_000);
+ polled.add(page(0, &[1; 990], 989), 990);
+ assert_eq!(next_page(&polled, usize::MAX, 1), Some(10));
+ }
+
+ #[test]
+ fn given_max_bytes_when_sizing_the_response_should_keep_it_within_one_frame() {
+ let frame = 8 * 1024 * 1024;
+ assert_eq!(response_bytes(52_428_800, frame), frame);
+ assert_eq!(response_bytes(1000, frame), 1000);
+ assert_eq!(response_bytes(-1, frame), 0);
+ }
+
+ #[test]
+ fn given_an_offset_when_placed_against_the_probe_should_follow_the_error_table() {
+ let found = probe(5, 10);
+ let probes = probes(&[(0, found)]);
+ let at = |partition, fetch_offset| position(&want(partition, fetch_offset), &probes);
+ assert_eq!(at(0, -1), Position::Refused(ERROR_OFFSET_OUT_OF_RANGE));
+ assert_eq!(
+ at(0, 6),
+ Position::Unsure {
+ index: 0,
+ offset: 6,
+ probe: found
+ },
+ "a poll decides, not a probe that may be old"
+ );
+ assert_eq!(at(0, 5), Position::CaughtUp(found));
+ assert_eq!(
+ at(0, 0),
+ Position::Behind {
+ index: 0,
+ offset: 0,
+ probe: found
+ }
+ );
+ assert_eq!(
+ at(1, 0),
+ Position::Refused(ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ );
+ assert_eq!(
+ at(-1, 0),
+ Position::Refused(ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ );
+ assert_eq!(
+ position(&want(0, 0), &vec![Err(ERROR_NOT_LEADER_OR_FOLLOWER)]),
+ Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER)
+ );
+ }
+
+ #[test]
+ fn given_a_partition_that_keeps_no_message_when_placed_should_poll_before_answering() {
+ let emptied = probe(42, 0);
+ let probes = probes(&[(0, emptied)]);
+ let at = |fetch_offset| position(&want(0, fetch_offset), &probes);
+ let unsure = |offset| Position::Unsure {
+ index: 0,
+ offset,
+ probe: emptied,
+ };
+ assert_eq!(at(42), Position::CaughtUp(emptied));
+ assert_eq!(at(10), unsure(10), "maybe a gap, maybe a stats miss");
+ assert_eq!(at(50), unsure(50), "maybe past the end, maybe a stats miss");
+ assert_eq!(at(-1), Position::Refused(ERROR_OFFSET_OUT_OF_RANGE));
+ assert!(
+ !ready(&[want(0, 10)], &[false], &probes, 1024, 1),
+ "no message kept proves no gap, so the wait goes on"
+ );
+ }
+
+ #[test]
+ fn given_an_offset_past_the_probe_when_waiting_should_let_a_newer_probe_decide() {
+ let probes = probes(&[(0, probe(5, 10)), (1, probe(42, 0))]);
+ assert!(
+ !ready(&[want(0, 6)], &[false], &probes, 1024, 1),
+ "maybe caught up on an old probe, so the wait goes on"
+ );
+ assert!(
+ !ready(&[want(1, 50)], &[false], &probes, 1024, 1),
+ "no message kept, past the probe: the same"
+ );
+ }
+
+ #[tokio::test]
+ async fn given_a_gateway_write_when_waiting_should_wake_for_that_topic() {
+ let board = ProbeBoard::default();
+ let mut writes = vec![board.writes("a"), board.writes("b"), board.writes("c")];
+ let quiet = tokio::time::timeout(Duration::from_millis(20), next_writes(&mut writes));
+ assert!(quiet.await.is_err(), "no write, no wake");
+ let before = Instant::now();
+ board.wrote("c");
+ board.wrote("a");
+ board.wrote("not waited on");
+ let after = Instant::now();
+ let written = tokio::time::timeout(Duration::from_secs(1), next_writes(&mut writes))
+ .await
+ .expect("the write wakes the wait");
+ let topics: Vec<usize> = written.iter().map(|&(topic, _)| topic).collect();
+ assert_eq!(topics, vec![0, 2]);
+ assert!(
+ written.iter().all(|&(_, at)| before <= at && at <= after),
+ "each with the time of its write, not of the wake"
+ );
+ let again = tokio::time::timeout(Duration::from_millis(20), next_writes(&mut writes));
+ assert!(again.await.is_err(), "each write is seen once");
+ }
+
+ #[test]
+ fn given_many_stale_topics_when_probing_should_load_the_oldest_first_up_to_the_cap() {
+ let now = Instant::now();
+ let snapshot = |age| {
+ Some(Arc::new(Snapshot {
+ started: now - Duration::from_millis(age),
+ probe: Ok(Arc::new(TopicProbe::default())),
+ }))
+ };
+ let snapshots = [snapshot(500), None, snapshot(10), snapshot(900)];
+ assert_eq!(
+ stale_topics(&snapshots, now - Duration::from_millis(100)),
+ vec![1, 3, 0],
+ "none first, then the oldest; a new one waits"
+ );
+ let many = vec![None; MAX_ROUND_TOPICS + 5];
+ assert_eq!(stale_topics(&many, now).len(), MAX_ROUND_TOPICS);
+ }
+
+ #[test]
+ fn given_a_partition_mid_reload_when_placed_should_answer_6() {
+ let reloading = PartitionProbe {
+ high_watermark: 1,
+ average_size: 10,
+ messages_count: 50,
+ };
+ let probes = probes(&[(0, reloading)]);
+ for fetch_offset in [0, 1, 500] {
+ assert_eq!(
+ position(&want(0, fetch_offset), &probes),
+ Position::Refused(ERROR_NOT_LEADER_OR_FOLLOWER)
+ );
+ }
+ }
+
+ #[test]
+ fn given_an_empty_poll_when_decided_should_trust_only_a_probe_taken_after_it() {
+ let before = probe(42, 0);
+ let empty = |high_watermark| Some(probe(high_watermark, 0));
+ let past = |high_watermark| EmptyPoll::OutOfRange { high_watermark };
+ assert_eq!(
+ empty_poll(10, before, empty(42)),
+ EmptyPoll::Unproven,
+ "no message kept: retention, or a replica that lost what its peers keep"
+ );
+ assert_eq!(empty_poll(10, before, empty(44)), EmptyPoll::Unproven);
+ assert_eq!(
+ empty_poll(10, before, Some(probe(42, 100))),
+ EmptyPoll::NotReady,
+ "messages kept, so the poll failed"
+ );
+ assert_eq!(empty_poll(10, before, None), EmptyPoll::Unproven);
+ assert_eq!(empty_poll(50, before, empty(42)), past(42));
+ assert_eq!(
+ empty_poll(43, before, empty(44)),
+ EmptyPoll::Unproven,
+ "a new message, not counted yet"
+ );
+ assert_eq!(empty_poll(44, before, empty(44)), EmptyPoll::Unproven);
+ }
+
+ #[test]
+ fn given_an_empty_first_poll_when_decided_should_not_read_a_stats_miss_as_a_gap() {
+ let before = probe(42, 100);
+ let miss = Some(probe(0, 0));
+ assert_eq!(
+ empty_poll(10, before, miss),
+ EmptyPoll::OutOfRange { high_watermark: 0 },
+ "(0, 0) is a stats miss or a load: the grace decides"
+ );
+ assert_eq!(
+ empty_poll(10, before, Some(probe(42, 100))),
+ EmptyPoll::NotReady,
+ "records sat there before the poll and still do"
+ );
+ assert_eq!(
+ empty_poll(10, before, Some(probe(42, 0))),
+ EmptyPoll::Unproven,
+ "no message kept is no gap: serve nothing"
+ );
+ let loading = PartitionProbe {
+ high_watermark: 1,
+ average_size: 10,
+ messages_count: 50,
+ };
+ assert_eq!(empty_poll(10, before, Some(loading)), EmptyPoll::NotReady);
+ assert_eq!(
+ empty_poll(50, probe(42, 100), Some(probe(42, 100))),
+ EmptyPoll::OutOfRange { high_watermark: 42 }
+ );
+ assert_eq!(
+ empty_poll(50, probe(42, 100), Some(probe(60, 100))),
+ EmptyPoll::Unproven,
+ "records came after the poll"
+ );
+ }
+
+ #[test]
+ fn given_a_partition_not_ready_when_it_stays_so_should_count_one_spell() {
+ let spells = Mutex::default();
+ let start = Instant::now();
+ let at = |seconds| start + Duration::from_secs(seconds);
+ assert_eq!(spell(&spells, "a", 0, at(0)), Duration::ZERO, "a new spell");
+ assert_eq!(spell(&spells, "a", 0, at(10)), Duration::from_secs(10));
+ assert_eq!(
+ spell(&spells, "a", 1, at(10)),
+ Duration::ZERO,
+ "per partition"
+ );
+ assert_eq!(spell(&spells, "a", 0, at(35)), Duration::from_secs(35));
+ assert!(
+ spell(&spells, "a", 0, at(40)) >= RELOAD_GRACE,
+ "past the grace: 1"
+ );
+ assert_eq!(
+ spell(&spells, "a", 0, at(40) + RELOAD_GRACE),
+ Duration::ZERO,
+ "a quiet grace ends the spell"
+ );
+ }
+
+ #[test]
+ fn given_bridge_errors_when_mapped_should_send_only_codes_the_consumer_knows() {
+ let table = [
+ (BridgeError::Timeout, ERROR_NOT_LEADER_OR_FOLLOWER),
+ (
+ BridgeError::Iggy(IggyError::TransientNotCommitted),
+ ERROR_NOT_LEADER_OR_FOLLOWER,
+ ),
+ (
+ BridgeError::Iggy(IggyError::Disconnected),
+ ERROR_NOT_LEADER_OR_FOLLOWER,
+ ),
+ (
+ BridgeError::Iggy(IggyError::Unauthenticated),
+ ERROR_NOT_LEADER_OR_FOLLOWER,
+ ),
+ (
+ BridgeError::Iggy(IggyError::Unauthorized),
+ ERROR_TOPIC_AUTHORIZATION_FAILED,
+ ),
+ (
+ BridgeError::Iggy(IggyError::TopicNameNotFound(
+ "orders".to_string(),
+ "kafka".to_string(),
+ )),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ),
+ (
+ BridgeError::PartitionOutOfRange {
+ topic: "orders".to_string(),
+ partition: 5,
+ partitions_count: 1,
+ },
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ),
+ (
+ BridgeError::InvalidKafkaTopicName {
+ kafka_topic: "bad name".to_string(),
+ reason: "space".to_string(),
+ },
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ),
+ (
+ BridgeError::Iggy(IggyError::InvalidCredentials),
+ ERROR_UNKNOWN_SERVER_ERROR,
+ ),
+ (
+ BridgeError::PartitionCountMismatch {
+ topic: "orders".to_string(),
+ existing: 1,
+ requested: 2,
+ },
+ ERROR_UNKNOWN_SERVER_ERROR,
+ ),
+ ];
+ for (error, expected) in table {
+ let code = fetch_code(&error);
+ assert_eq!(code, expected, "{error}");
+ assert!(
+ CONSUMER_CODES.contains(&code),
+ "{code} makes the Java consumer throw"
+ );
+ }
+ }
+
+ #[test]
+ fn given_session_fields_when_checked_should_accept_only_a_full_request() {
+ let with = |session_id, session_epoch| {
+ request(&[])
+ .with_session_id(session_id)
+ .with_session_epoch(session_epoch)
+ };
+ assert!(is_full(6, &with(7, 3)), "no sessions before v7");
+ assert!(is_full(7, &with(0, INITIAL_EPOCH)));
+ assert!(is_full(12, &with(0, FINAL_EPOCH)));
+ assert!(
+ is_full(12, &with(5, INITIAL_EPOCH)),
+ "closes session 5 and asks for a new one"
+ );
+ assert!(!is_full(7, &with(5, 1)));
+ assert!(!is_full(12, &with(5, 9)));
+ assert!(!is_full(12, &with(0, -2)));
+ }
+
+ #[test]
+ fn given_a_record_past_every_limit_when_owed_should_still_return_it() {
+ let messages = [stored(0, &[b'a'; 1024]), stored(1, &[b'b'; 1024])];
+
+ let owed = encode_records(&messages, 0, 10, true).unwrap();
+ assert_eq!(offsets(owed), vec![0], "exactly the first record");
+
+ let not_owed = encode_records(&messages, 0, 10, false).unwrap();
+ assert!(not_owed.is_empty(), "no records rather than a cut batch");
+ }
+
+ #[test]
+ fn given_a_budget_when_encoding_should_stop_before_passing_it() {
+ let messages: Vec<IggyMessage> =
+ (0..10).map(|offset| stored(offset, &[b'v'; 100])).collect();
+
+ let batch = encode_records(&messages, 0, 500, false).unwrap();
+ assert!(batch.len() <= 500, "{} bytes", batch.len());
+ let served = offsets(batch);
+ assert!(!served.is_empty() && served.len() < 10);
+ assert_eq!(served, (0..).take(served.len()).collect::<Vec<i64>>());
+ }
+
+ #[test]
+ fn given_messages_below_the_fetch_offset_when_encoding_should_skip_them() {
+ let messages = [stored(3, b"a"), stored(4, b"b"), stored(5, b"c")];
+ let batch = encode_records(&messages, 4, usize::MAX, false).unwrap();
+ assert_eq!(offsets(batch), vec![4, 5]);
+ }
+
+ #[test]
+ fn given_an_unmappable_message_when_it_comes_first_should_fail_the_partition() {
+ let messages = [unmappable(0), stored(1, b"b")];
+ assert!(matches!(
+ encode_records(&messages, 0, usize::MAX, true),
+ Err(RecordCodecError::MappingVersion(_))
+ ));
+ }
+
+ #[test]
+ fn given_an_unmappable_message_after_others_should_serve_the_ones_before_it() {
+ let messages = [
+ stored(0, b"a"),
+ stored(1, b"b"),
+ unmappable(2),
+ stored(3, b"d"),
+ ];
+ let batch = encode_records(&messages, 0, usize::MAX, true).unwrap();
+ assert_eq!(
+ offsets(batch),
+ vec![0, 1],
+ "the next Fetch starts at the bad one"
+ );
+ }
+
+ #[test]
+ fn given_a_partition_named_twice_when_answered_should_answer_it_once() {
+ let request = request(&[("a", &[0, 1]), ("b", &[0]), ("a", &[1, 2])]);
+ let (wanted, _) = wanted(&request);
+ let named: Vec<(&str, i32)> = wanted
+ .iter()
+ .map(|want| (want.topic, want.partition))
+ .collect();
+ assert_eq!(named, vec![("a", 0), ("a", 1), ("b", 0), ("a", 2)]);
+
+ let answers = wanted
+ .iter()
+ .map(|want| served(want.partition, 0, Bytes::new()))
+ .collect();
+ let responses = group(&request, &wanted, answers);
+ let grouped: Vec<(&str, Vec<i32>)> = responses
+ .iter()
+ .map(|topic| {
+ (
+ topic.topic.as_str(),
+ topic.partitions.iter().map(|p| p.partition_index).collect(),
+ )
+ })
+ .collect();
+ assert_eq!(
+ grouped,
+ vec![("a", vec![0, 1]), ("b", vec![0]), ("a", vec![2])],
+ "under their own request entries"
+ );
+ }
+
+ #[test]
+ fn given_repeated_topics_when_listed_should_probe_each_once_and_skip_negative_indexes() {
+ let request = request(&[("a", &[2, -1, 0]), ("b", &[-1]), ("c", &[1]), ("a", &[1])]);
+ let (wanted, topics) = wanted(&request);
+ assert_eq!(topics, vec!["a", "c"], "b has no index to probe");
+ let at = |topic, index| Ok(ProbeAt { topic, index });
+ let negative = Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ assert_eq!(
+ wanted.iter().map(|want| want.probe_at).collect::<Vec<_>>(),
+ vec![at(0, 2), negative, at(0, 0), negative, at(1, 1), at(0, 1)]
+ );
+ }
+
+ #[test]
+ fn given_a_name_kafka_refuses_when_listed_should_answer_3_without_a_probe() {
+ let request = request(&[("bad name", &[0]), ("a", &[0])]);
+ let (wanted, topics) = wanted(&request);
+ assert_eq!(topics, vec!["a"]);
+ assert_eq!(wanted[0].probe_at, Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION));
+ assert_eq!(wanted[1].probe_at, Ok(ProbeAt { topic: 0, index: 0 }));
+ }
+
+ #[test]
+ fn given_a_served_partition_when_built_should_report_no_transactions() {
+ let answer = served(3, 42, Bytes::new());
+ assert_eq!(answer.error_code, ERROR_NONE);
+ assert_eq!(answer.high_watermark, 42);
+ assert_eq!(answer.last_stable_offset, 42);
+ assert_eq!(answer.log_start_offset, UNKNOWN_OFFSET);
+ assert_eq!(answer.aborted_transactions, None);
+
+ let refused = refused(3, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ assert_eq!(refused.high_watermark, UNKNOWN_OFFSET);
+ assert_eq!(refused.last_stable_offset, UNKNOWN_OFFSET);
+ }
+
+ #[test]
+ fn given_a_minus_one_when_asked_again_should_hold_the_partition_until_the_wait_ends() {
+ let stuck = StuckOffsets::default();
+ let now = Instant::now();
+ let until = now + Duration::from_millis(500);
+ assert!(hold(&stuck, "a", 0, 7, until), "a new stop");
+ assert!(held(&stuck, "a", 0, 7, now));
+ assert!(!held(&stuck, "a", 0, 7, until), "the hold ended");
+ assert!(!held(&stuck, "a", 0, 8, now), "another offset");
+ assert!(!held(&stuck, "a", 1, 7, now), "another partition");
+ assert!(!held(&stuck, "b", 0, 7, now), "another topic");
+ assert!(!hold(&stuck, "a", 0, 7, until), "the same stop");
+ assert!(
+ hold(&stuck, "a", 0, 9, until),
+ "moved on, then stopped again"
+ );
+ }
+
+ #[test]
+ fn given_a_6_or_a_minus_one_in_hand_when_the_request_allows_a_wait_should_wait_it_out() {
+ let request = request(&[]).with_max_wait_ms(500).with_min_bytes(1);
+ for code in [ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_UNKNOWN_SERVER_ERROR] {
+ let stuck = refused(0, code);
+ assert!(waits(&request, std::slice::from_ref(&stuck)), "{code}");
+ assert!(
+ waits(&request, &[stuck.clone(), served(1, 5, Bytes::new())]),
+ "{code}"
+ );
+ assert!(
+ !waits(
+ &request,
+ &[stuck, served(1, 5, Bytes::from_static(b"batch"))]
+ ),
+ "records in hand"
+ );
+ }
+
+ let wanted = [want(0, 10), want(1, 5)];
+ let probes = |second| probes(&[(0, probe(20, 100)), (1, probe(second, 100))]);
+ let held = [true, false];
+ assert!(
+ !ready(&wanted, &held, &probes(5), 1024, 1),
+ "a held partition ends no wait"
+ );
+ assert!(ready(&wanted, &held, &probes(6), 1024, 1));
+ }
+
+ #[test]
+ fn given_records_below_min_bytes_when_the_request_allows_a_wait_should_wait() {
+ let request = request(&[]).with_max_wait_ms(500).with_min_bytes(100);
+ let small = served(0, 5, Bytes::from_static(b"batch"));
+ assert!(waits(&request, std::slice::from_ref(&small)));
+ let enough = served(0, 5, Bytes::from(vec![0; 100]));
+ assert!(!waits(&request, &[small, enough]), "min_bytes reached");
+ assert!(
+ !waits(
+ &request,
+ &[
+ served(0, 5, Bytes::new()),
+ refused(1, ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ ]
+ ),
+ "an error in hand"
+ );
+ }
+
+ #[test]
+ fn given_nothing_found_when_the_request_allows_a_wait_should_wait() {
+ let empty = [served(0, 5, Bytes::new())];
+ let request = request(&[]).with_max_wait_ms(500).with_min_bytes(1);
+ assert!(waits(&request, &empty));
+ assert!(
+ !waits(&request.clone().with_max_wait_ms(0), &empty),
+ "no wait asked for"
+ );
+ assert!(
+ !waits(&request.clone().with_min_bytes(0), &empty),
+ "zero bytes are already there"
+ );
+ assert!(!waits(&request, &[]), "no partition to wait on");
+ assert!(
+ !waits(&request, &[served(0, 5, Bytes::from_static(b"batch"))]),
+ "records in hand"
+ );
+ assert!(
+ !waits(&request, &[refused(0, ERROR_UNKNOWN_TOPIC_OR_PARTITION)]),
+ "an error in hand"
+ );
+ }
+
+ #[test]
+ fn given_new_messages_when_estimated_should_count_at_most_the_budget() {
+ assert_eq!(bytes_waiting(8, probe(10, 100), 1024), 200);
+ assert_eq!(
+ bytes_waiting(0, probe(10, 100), 500),
+ 500,
+ "capped at the budget"
+ );
+ assert_eq!(
+ bytes_waiting(0, probe(10, 0), 500),
+ 0,
+ "nothing retained to count"
+ );
+ }
+
+ #[test]
+ fn given_new_probes_when_waiting_should_stop_on_enough_bytes_or_a_failure() {
+ let wanted = [want(0, 10)];
+ let frame = 8 * 1024 * 1024;
+ let at = |high_watermark| probes(&[(0, probe(high_watermark, 100))]);
+
+ assert!(
+ !ready(&wanted, &[false], &at(10), frame, 1),
+ "still caught up"
+ );
+ assert!(
+ ready(&wanted, &[false], &at(11), frame, 1),
+ "one new message"
+ );
+ assert!(
+ !ready(&wanted, &[false], &at(11), frame, 1000),
+ "short of min_bytes"
+ );
+ assert!(
+ ready(
+ &wanted,
+ &[false],
+ &vec![Err(ERROR_NOT_LEADER_OR_FOLLOWER)],
+ frame,
+ 1
+ ),
+ "a failed probe"
+ );
+ assert!(
+ ready(&wanted, &[false], &probes(&[]), frame, 1),
+ "the partition is gone"
+ );
+ }
+
+ #[test]
+ fn given_many_topics_when_waiting_should_probe_each_within_a_few_rounds() {
+ let mut seen = HashSet::new();
+ for turn in 0..3 {
+ let topics: Vec<usize> = round(turn, 250).collect();
+ assert_eq!(topics.len(), MAX_ROUND_TOPICS);
+ seen.extend(topics);
+ }
+ assert_eq!(seen.len(), 250, "three rounds cover 250 topics");
+
+ let mut few: Vec<usize> = round(7, 3).collect();
+ few.sort_unstable();
+ assert_eq!(few, vec![0, 1, 2], "each round covers a short list");
+ assert_eq!(round(0, 0).count(), 0);
+ }
+
+ #[test]
+ fn given_an_empty_poll_when_the_commit_ends_before_the_offset_should_read_as_caught_up() {
+ assert!(caught_up(42, 41));
+ assert!(!caught_up(42, 40), "a message may still wait");
+ assert!(!caught_up(42, 42), "past the end");
+ assert!(
+ !caught_up(1, 0),
+ "0 comes from an empty partition or a failed read too"
+ );
+ assert!(!caught_up(0, 0));
+ }
+
+ #[test]
+ fn given_a_short_wait_when_a_minus_one_holds_should_hold_one_probe_interval_at_least() {
+ let now = Instant::now();
+ assert_eq!(hold_until(now, Duration::ZERO), now + PROBE_INTERVAL);
+ assert_eq!(
+ hold_until(now, Duration::from_millis(500)),
+ now + Duration::from_millis(500)
+ );
+ }
+
+ #[test]
+ fn given_max_wait_when_converted_should_stay_between_zero_and_the_cap() {
+ assert_eq!(wait_time(500), Duration::from_millis(500));
+ assert_eq!(wait_time(-1), Duration::ZERO);
+ assert_eq!(
+ wait_time(i32::MAX),
+ Duration::from_secs(18),
+ "time left to read before the request deadline"
+ );
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/list_offsets.rs b/gateways/kafka/src/protocol/handlers/list_offsets.rs
index ef9b2d5b4..60221f406 100644
--- a/gateways/kafka/src/protocol/handlers/list_offsets.rs
+++ b/gateways/kafka/src/protocol/handlers/list_offsets.rs
@@ -18,7 +18,6 @@
//! `ListOffsets` (API key 2).
use std::collections::{HashMap, HashSet};
-use std::time::Duration;
use bytes::Bytes;
use kafka_protocol::messages::list_offsets_request::{ListOffsetsPartition, ListOffsetsTopic};
@@ -31,15 +30,12 @@ use tokio::time::Instant;
use crate::bridge::{BridgeError, IggyBridge};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_NOT_LEADER_OR_FOLLOWER,
- ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
- ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
+ API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_REQUEST_TIMED_OUT,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, GatewayState,
+ HandleOutcome, REQUEST_DEADLINE, UNKNOWN_OFFSET,
};
use crate::protocol::bounds_guard::validate_list_offsets_shape;
-use crate::protocol::handlers::{
- decode_guarded, encode_message, handle_versioned_request, is_supported_version,
- respond_or_close, unsupported_version_response,
-};
+use crate::protocol::handlers::{decode_guarded, decode_request, encode_message, respond_or_close};
pub const RANGE: ApiVersionRange = ApiVersionRange {
api_key: API_KEY_LIST_OFFSETS,
@@ -62,78 +58,45 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
/// its own within a couple of retries, rather than resending the same oversized request forever.
const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
-/// Wall-clock ceiling for one request's aggregate bridge work.
-///
-/// `ListOffsets` carries no `timeout_ms` field in any version this gateway supports (that field
-/// is v10+; [`RANGE`] tops out at v6) - unlike `CreateTopics`, there is no client-supplied value
-/// to honor here, so this is a fixed ceiling instead. Sized well above one `high_watermarks`
-/// call's own `REQUEST_TIMEOUT` (15s, bridge-internal) so a single slow-but-alive call is not the
-/// common trigger, while still bounding the sum across up to [`MAX_BRIDGE_BACKED_TOPICS`] calls -
-/// without this, a large batch against a struggling bridge could hold the shared client for
-/// `MAX_BRIDGE_BACKED_TOPICS * 15s`, not just one call's worth.
-///
-/// Applied per call, not once around the whole batch: [`resolve_all_topics`] checks it before
-/// starting each topic's `high_watermarks` call and wraps the call itself in
-/// [`tokio::time::timeout_at`] against the same instant, so a topic already resolved when the
-/// deadline arrives keeps its real answer and only the not-yet-started ones fall back to
-/// [`ERROR_REQUEST_TIMED_OUT`].
-const REQUEST_DEADLINE: Duration = Duration::from_secs(20);
-
/// KIP-79 sentinel: the offset of the next message that would be produced.
const LATEST_TIMESTAMP: i64 = -1;
/// KIP-79 sentinel: the offset of the first message still retained.
const EARLIEST_TIMESTAMP: i64 = -2;
-/// Placeholder offset/timestamp for a partition result that carries an error - matches real
-/// Kafka's own convention on the error path.
-const NO_OFFSET: i64 = -1;
/// [`IggyBridge::high_watermarks`]'s return type, spelled once for [`resolve_one_partition`].
type HighWatermarksResult =
core::result::Result<Vec<(u32, core::result::Result<i64, BridgeError>)>, BridgeError>;
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
- let Some(bridge) = &state.bridge else {
- return handle_versioned_request(
- API_KEY_LIST_OFFSETS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<ListOffsetsRequest>(v, b, |v, b| {
- validate_list_offsets_shape(v, b, state.max_frame_size)
- })
- },
- encode_response,
- encode_error_response,
- "ListOffsets",
- );
- };
-
- if !is_supported_version(API_KEY_LIST_OFFSETS, api_version) {
- return unsupported_version_response(API_KEY_LIST_OFFSETS, api_version, |version| {
- encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
- });
- }
-
- let req = match decode_guarded::<ListOffsetsRequest>(api_version, body, |v, b| {
- validate_list_offsets_shape(v, b, state.max_frame_size)
- }) {
+ let decoded = decode_request(
+ API_KEY_LIST_OFFSETS,
+ api_version,
+ body,
+ |v, b| decode(state, v, b),
+ encode_error_response,
+ "ListOffsets",
+ );
+ let req = match decoded {
Ok(req) => req,
- Err(error) => {
- // debug!, not warn!: attacker-controlled, not operator-actionable.
- tracing::debug!(%error, "Failed to decode ListOffsets request");
- return respond_or_close(
- encode_error_response(api_version, ERROR_INVALID_REQUEST),
- "ListOffsets",
- );
- }
+ Err(outcome) => return outcome,
+ };
+ let Some(bridge) = &state.bridge else {
+ return respond_or_close(encode_response(api_version, &req), "ListOffsets");
};
+ // No `timeout_ms` below v10, so the ceiling is fixed. It bounds the sum of the topic calls.
let deadline = Instant::now() + REQUEST_DEADLINE;
let topics = resolve_all_topics(bridge, &req.topics, deadline).await;
let resp = ListOffsetsResponse::default().with_topics(topics);
respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
}
+fn decode(state: &GatewayState, version: i16, body: Bytes) -> Result<ListOffsetsRequest> {
+ decode_guarded::<ListOffsetsRequest>(version, body, |v, b| {
+ validate_list_offsets_shape(v, b, state.max_frame_size)
+ })
+}
+
/// One topic's bridge-lookup outcome, decided once per distinct name in [`resolve_all_topics`]
/// and reused for every partition of that topic in [`resolve_one_partition`].
enum TopicLookup {
@@ -144,7 +107,8 @@ enum TopicLookup {
Watermarks(HighWatermarksResult),
/// Beyond [`MAX_BRIDGE_BACKED_TOPICS`], or the deadline elapsed before this topic's turn - no
/// call was made. Answered [`ERROR_REQUEST_TIMED_OUT`] (retriable) rather than
- /// [`ERROR_INVALID_REQUEST`] so a client's own per-topic retry narrows the batch on its own.
+ /// [`ERROR_INVALID_REQUEST`](crate::protocol::api::ERROR_INVALID_REQUEST) so a client's own
+ /// per-topic retry narrows the batch on its own.
NotAttempted,
}
@@ -340,9 +304,9 @@ fn resolve_one_partition(
// Real only for a partition retention has never trimmed: Iggy tracks no rolling
// low-watermark distinct from partition creation, so a `0` here for an older,
// already-trimmed partition names a log-start offset that no longer exists - a real
- // consumer with `auto.offset.reset=earliest` would seek into a hole. Harmless *today*
- // only because Fetch (`#3536`) is still a stub - nothing yet reads at the offset this
- // returns. Not fixable client-side; needs the bridge to expose a real start offset.
+ // consumer with `auto.offset.reset=earliest` would seek into a hole. Fetch there reads
+ // from the oldest kept message, so the consumer skips the gap. Not fixable client-side;
+ // needs the bridge to expose a real start offset.
EARLIEST_TIMESTAMP => offset_response(requested.partition_index, 0),
// Non-retriable, unlike ERROR_UNKNOWN_SERVER_ERROR: a Java client resolves this
// immediately instead of retrying the request until its own default.api.timeout.ms.
@@ -364,8 +328,8 @@ fn error_response(partition: i32, error_code: i16) -> ListOffsetsPartitionRespon
ListOffsetsPartitionResponse::default()
.with_partition_index(partition)
.with_error_code(error_code)
- .with_timestamp(NO_OFFSET)
- .with_offset(NO_OFFSET)
+ .with_timestamp(UNKNOWN_OFFSET)
+ .with_offset(UNKNOWN_OFFSET)
}
/// Well-formed `ListOffsets` response with a single placeholder topic/partition.
@@ -462,7 +426,7 @@ mod tests {
let lookup = TopicLookup::Watermarks(Ok(ok_watermarks(&[(0, 42)])));
let resp = resolve_one_partition(&partition(0, 1_700_000_000_000), &lookup);
assert_eq!(resp.error_code, ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT);
- assert_eq!(resp.offset, NO_OFFSET);
+ assert_eq!(resp.offset, UNKNOWN_OFFSET);
}
#[test]
@@ -481,7 +445,7 @@ mod tests {
let resp =
resolve_one_partition(&partition(0, LATEST_TIMESTAMP), &TopicLookup::NotAttempted);
assert_eq!(resp.error_code, ERROR_REQUEST_TIMED_OUT);
- assert_eq!(resp.offset, NO_OFFSET);
+ assert_eq!(resp.offset, UNKNOWN_OFFSET);
}
#[test]
diff --git a/gateways/kafka/src/protocol/handlers/metadata.rs b/gateways/kafka/src/protocol/handlers/metadata.rs
index 8e1cd5dda..69685d9ac 100644
--- a/gateways/kafka/src/protocol/handlers/metadata.rs
+++ b/gateways/kafka/src/protocol/handlers/metadata.rs
@@ -18,7 +18,6 @@
//! Metadata (API key 3).
use std::collections::HashSet;
-use std::time::Duration;
use bytes::Bytes;
use kafka_protocol::messages::metadata_response::{
@@ -31,8 +30,8 @@ use crate::bridge::IggyBridge;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
API_KEY_METADATA, ApiVersionRange, BrokerAdvertise, ERROR_NONE, ERROR_REQUEST_TIMED_OUT,
- ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, is_supported_version,
- supported_max_version,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, REQUEST_DEADLINE,
+ is_supported_version, supported_max_version,
};
use crate::protocol::bounds_guard::validate_metadata_shape;
use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
@@ -43,15 +42,6 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
max_version: 9,
};
-/// Wall-clock ceiling for a named-lookup request's aggregate bridge work.
-///
-/// `Metadata` carries no `timeout_ms` field in any version (unlike `CreateTopics`), so this is a
-/// fixed ceiling, not a client-honored one - sized well above one `get_topics` call's own
-/// `REQUEST_TIMEOUT` (15s, bridge-internal) so a single slow-but-alive stream lookup is not the
-/// common trigger, while still bounding the sum across every distinct stream
-/// [`IggyBridge::get_kafka_topics`] ends up calling for this request.
-const REQUEST_DEADLINE: Duration = Duration::from_secs(20);
-
/// Above this many total partitions in an all-topics response, `encode_real_response` runs on a
/// blocking-pool thread instead of inline (see the call site). Conservative relative to the one
/// measured data point available (20-54ms at 131,000 partitions) rather than a profiled
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 3b3d47a6b..257987f6f 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -40,6 +40,7 @@ pub mod sync_group;
use bytes::{Buf, Bytes, BytesMut};
use kafka_protocol::messages::TransactionalId;
use kafka_protocol::protocol::{Decodable, Encodable};
+use tokio::runtime::{Handle, RuntimeFlavor};
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
@@ -50,6 +51,23 @@ use crate::protocol::api::{
is_supported_version, supported_max_version,
};
+/// Record encodes and decodes of this many bytes or more run off the async worker.
+pub(crate) const CODEC_OFF_WORKER_BYTES: usize = 64 * 1024;
+/// A plain copy costs less per byte, so the handoff pays off only from this size.
+pub(crate) const COPY_OFF_WORKER_BYTES: usize = 1024 * 1024;
+
+/// Runs `work`. When `heavy`, other tasks move to another worker first, so they keep running. A
+/// current-thread runtime runs it in place.
+pub(crate) fn off_worker<T>(heavy: bool, work: impl FnOnce() -> T) -> T {
+ let multi_thread = Handle::try_current()
+ .is_ok_and(|handle| handle.runtime_flavor() == RuntimeFlavor::MultiThread);
+ if heavy && multi_thread {
+ tokio::task::block_in_place(work)
+ } else {
+ work()
+ }
+}
+
/// Routes one decoded request body to the module that owns its API key.
///
/// An unknown key closes the connection: no response schema exists for it, so any body sent
@@ -155,22 +173,35 @@ pub(crate) fn handle_versioned_request<T>(
encode_err: impl Fn(i16, i16) -> Result<Bytes>,
api_name: &str,
) -> HandleOutcome {
- if is_supported_version(api_key, api_version) {
- match decode(api_version, body) {
- Ok(req) => respond_or_close(encode_ok(api_version, &req), api_name),
- Err(error) => {
- // debug!, not warn!: attacker-controlled, not operator-actionable.
- tracing::debug!(%error, "Failed to decode {api_name} request");
- respond_or_close(encode_err(api_version, ERROR_INVALID_REQUEST), api_name)
- }
- }
- } else {
- unsupported_version_response(api_key, api_version, |version| {
- encode_err(version, ERROR_UNSUPPORTED_VERSION)
- })
+ match decode_request(api_key, api_version, body, decode, encode_err, api_name) {
+ Ok(req) => respond_or_close(encode_ok(api_version, &req), api_name),
+ Err(outcome) => outcome,
}
}
+/// The decoded request, or the answer to send when the version or the body is bad.
+pub(crate) fn decode_request<T>(
+ api_key: i16,
+ api_version: i16,
+ body: Bytes,
+ decode: impl FnOnce(i16, Bytes) -> Result<T>,
+ encode_err: impl Fn(i16, i16) -> Result<Bytes>,
+ api_name: &str,
+) -> core::result::Result<T, HandleOutcome> {
+ if !is_supported_version(api_key, api_version) {
+ return Err(unsupported_version_response(
+ api_key,
+ api_version,
+ |version| encode_err(version, ERROR_UNSUPPORTED_VERSION),
+ ));
+ }
+ decode(api_version, body).map_err(|error| {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode {api_name} request");
+ respond_or_close(encode_err(api_version, ERROR_INVALID_REQUEST), api_name)
+ })
+}
+
/// Unsupported-version policy for APIs whose encoders only implement up to
/// `ApiVersionRange::max_version`.
///
@@ -201,3 +232,23 @@ pub(crate) fn unsupported_version_response(
}
respond_or_close(encode(api_version), "unsupported-version")
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_no_runtime_when_work_is_large_should_run_it_in_place() {
+ assert_eq!(off_worker(true, || 7), 7);
+ }
+
+ #[tokio::test]
+ async fn given_a_current_thread_runtime_when_work_is_large_should_run_it_in_place() {
+ assert_eq!(off_worker(true, || 7), 7);
+ }
+
+ #[tokio::test(flavor = "multi_thread")]
+ async fn given_a_multi_thread_runtime_when_work_is_large_should_run_it() {
+ assert_eq!(off_worker(true, || 7), 7);
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs
index e64e22ac0..a18b9c388 100644
--- a/gateways/kafka/src/protocol/handlers/produce.rs
+++ b/gateways/kafka/src/protocol/handlers/produce.rs
@@ -25,7 +25,6 @@ use kafka_protocol::messages::produce_request::PartitionProduceData;
use kafka_protocol::messages::produce_response::{PartitionProduceResponse, TopicProduceResponse};
use kafka_protocol::messages::{ProduceRequest, ProduceResponse};
use kafka_protocol::protocol::StrBytes;
-use tokio::runtime::{Handle, RuntimeFlavor};
use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
use crate::error::Result;
@@ -34,13 +33,14 @@ use crate::protocol::api::{
ERROR_MESSAGE_TOO_LARGE, ERROR_NONE, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_REQUEST_TIMED_OUT,
ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
ERROR_UNSUPPORTED_COMPRESSION_TYPE, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
- supported_max_version,
+ REQUEST_DEADLINE, UNKNOWN_OFFSET, supported_max_version,
};
use crate::protocol::bounds_guard::validate_produce_shape;
use crate::protocol::handlers::{
- decode_guarded, encode_message, is_transactional, respond_or_close,
- unsupported_version_response,
+ CODEC_OFF_WORKER_BYTES, decode_guarded, encode_message, is_transactional, off_worker,
+ respond_or_close, unsupported_version_response,
};
+use crate::protocol::probe_board::ProbeBoard;
use crate::records::{
Allowance, DecompressionBudget, RecordCodecError, TimestampWindow, Zstd, decode_batch,
is_compressed, to_iggy,
@@ -60,18 +60,6 @@ const ACKS_ALL: i16 = -1;
/// First Produce version that may carry zstd batches.
const ZSTD_MIN_VERSION: i16 = 7;
-/// 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, 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.
-const MAX_REQUEST_DEADLINE: Duration = Duration::from_secs(20);
-
/// Frame bytes per record slot. Caps one partition at 131,072 slots, about 40 MB, at the default
/// 8 MiB frame.
///
@@ -84,9 +72,6 @@ const FRAME_BYTES_PER_RECORD: usize = 64;
/// answers 6.
const REQUEST_ALLOWANCES: usize = 8;
-/// Blobs this large, or compressed ones, plan off the async worker.
-const BLOCKING_PLAN_BYTES: usize = 64 * 1024;
-
const RESPONSE_BASE_BYTES: usize = 512;
/// Produce is the only request the wire protocol allows to go unanswered
@@ -193,7 +178,8 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
} else {
Zstd::Refused
};
- let responses = write_request(bridge, &request, budget, zstd, deadline).await;
+ let responses =
+ write_request(bridge, &state.probe_board, &request, budget, zstd, deadline).await;
// Kafka closes an `acks=0` connection on any error, so the client refreshes metadata.
if request.acks == ACKS_NONE {
@@ -233,7 +219,7 @@ impl Deadline {
let wait = request_timeout(timeout_ms);
Self {
at: tokio::time::Instant::now() + wait,
- ceiling: wait == MAX_REQUEST_DEADLINE,
+ ceiling: wait == REQUEST_DEADLINE,
}
}
@@ -276,6 +262,7 @@ impl From<RecordCodecError> for Refusal {
/// just queue on its stream mutex (see the README's "Concurrency ceiling").
async fn write_request(
bridge: &IggyBridge,
+ board: &ProbeBoard,
request: &ProduceRequest,
budget: DecompressionBudget,
zstd: Zstd,
@@ -287,6 +274,7 @@ async fn write_request(
let name = topic.name.as_str();
let target = bridge.topic_target(name);
let mut responses = Vec::with_capacity(topic.partition_data.len());
+ let mut wrote = false;
for partition in &topic.partition_data {
let outcome = match &target {
// Skip the decode. The answer is 7 either way.
@@ -302,10 +290,14 @@ async fn write_request(
}
Err(error) => Err(refusal_code(name, error).into()),
};
+ wrote |= outcome.is_ok();
responses.push(partition_outcome(partition.index, outcome));
// Planning is sync CPU work. Let other connections on this worker run.
tokio::task::yield_now().await;
}
+ if wrote {
+ board.wrote(name);
+ }
topics.push(
TopicProduceResponse::default()
.with_name(topic.name.clone())
@@ -330,12 +322,8 @@ fn plan(
&& partition
.records
.as_ref()
- .is_some_and(|blob| blob.len() >= BLOCKING_PLAN_BYTES || is_compressed(blob));
- if heavy && Handle::current().runtime_flavor() == RuntimeFlavor::MultiThread {
- tokio::task::block_in_place(run)
- } else {
- run()
- }
+ .is_some_and(|blob| blob.len() >= CODEC_OFF_WORKER_BYTES || is_compressed(blob));
+ off_worker(heavy, run)
}
fn partition_outcome(
@@ -416,8 +404,8 @@ 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)
+ .map_or(REQUEST_DEADLINE, Duration::from_millis)
+ .min(REQUEST_DEADLINE)
}
/// Sends one partition's messages, or answers 7 once the deadline has passed.
@@ -574,6 +562,7 @@ fn partition_response(index: i32, error_code: i16) -> PartitionProduceResponse {
.with_index(index)
.with_error_code(error_code)
.with_base_offset(UNKNOWN_OFFSET)
+ // The real log start costs a `get_topic` per request, and no producer reads it.
.with_log_start_offset(UNKNOWN_OFFSET)
}
@@ -867,18 +856,18 @@ mod tests {
assert_eq!(request_timeout(5_000), Duration::from_secs(5));
assert_eq!(
request_timeout(30_000),
- MAX_REQUEST_DEADLINE,
+ 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);
+ assert_eq!(request_timeout(i32::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);
+ assert_eq!(request_timeout(0), REQUEST_DEADLINE);
+ assert_eq!(request_timeout(-1), REQUEST_DEADLINE);
+ assert_eq!(request_timeout(i32::MIN), REQUEST_DEADLINE);
}
#[test]
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/protocol/mod.rs
index 34d8ed498..a32b2d092 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/protocol/mod.rs
@@ -20,4 +20,5 @@ pub mod api;
pub mod bounds_guard;
pub mod handlers;
pub mod header;
+pub(crate) mod probe_board;
pub mod sasl;
diff --git a/gateways/kafka/src/protocol/probe_board.rs b/gateways/kafka/src/protocol/probe_board.rs
new file mode 100644
index 000000000..73e45aca4
--- /dev/null
+++ b/gateways/kafka/src/protocol/probe_board.rs
@@ -0,0 +1,368 @@
+// 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.
+
+//! The newest probe of each Kafka topic, shared by every Fetch.
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use tokio::sync::{Mutex as AsyncMutex, watch};
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::TopicProbe;
+
+/// How long a probe counts as new, and how often a waiting Fetch asks for one.
+pub const PROBE_INTERVAL: Duration = Duration::from_millis(100);
+
+/// A topic's probe, or the Kafka code the probe failed with.
+pub type TopicProbed = Result<Arc<TopicProbe>, i16>;
+
+/// The board sweeps no map smaller than this.
+const MIN_SWEEP: usize = 1024;
+
+/// The oldest start a shared probe can have and still count as new at `now`.
+pub fn recent(now: Instant) -> Instant {
+ now.checked_sub(PROBE_INTERVAL).unwrap_or(now)
+}
+
+/// A probe, and when its `get_topic` started.
+#[derive(Debug)]
+pub struct Snapshot {
+ pub started: Instant,
+ pub probe: TopicProbed,
+}
+
+/// The newest probe of each Kafka topic, shared by every Fetch.
+///
+/// One refresh runs at a time per topic, in its own task, so a Fetch that gives up leaves no call
+/// behind in the SDK queue.
+#[derive(Default)]
+pub struct ProbeBoard {
+ topics: Mutex<Topics>,
+}
+
+/// Every topic on the board, and the size at which the next insert sweeps first.
+#[derive(Default)]
+struct Topics {
+ by_name: HashMap<String, Arc<Topic>>,
+ sweep_at: usize,
+}
+
+impl Topics {
+ /// Forgets each topic that no Fetch and no refresh holds, and whose probe is too old to share.
+ /// Any client can name topics, so the board must not keep them all.
+ fn sweep(&mut self, now: Instant) {
+ let since = recent(now);
+ self.by_name
+ .retain(|_, topic| Arc::strong_count(topic) > 1 || topic.newer_than(since));
+ // Twice what is left, so sweeps cost O(1) per insert.
+ self.sweep_at = self.by_name.len().saturating_mul(2).max(MIN_SWEEP);
+ }
+}
+
+#[derive(Default)]
+struct Topic {
+ newest: watch::Sender<Option<Arc<Snapshot>>>,
+ /// Held by the refresh that runs.
+ refresh: Arc<AsyncMutex<()>>,
+ /// When the last gateway write ended, so a Fetch that waits can probe at once.
+ written: watch::Sender<Option<Instant>>,
+}
+
+impl Topic {
+ /// Whether its newest probe started at `since` or later.
+ fn newer_than(&self, since: Instant) -> bool {
+ self.newest
+ .borrow()
+ .as_ref()
+ .is_some_and(|snapshot| snapshot.started >= since)
+ }
+}
+
+impl ProbeBoard {
+ /// A probe of `topic` that started at `since` or later. `None` if none comes by `deadline`.
+ ///
+ /// Runs `load` in a task when no refresh of `topic` runs.
+ pub async fn probe<F, Fut>(
+ &self,
+ topic: &str,
+ since: Instant,
+ deadline: Instant,
+ load: F,
+ ) -> Option<Arc<Snapshot>>
+ where
+ F: FnOnce() -> Fut,
+ Fut: Future<Output = TopicProbed> + Send + 'static,
+ {
+ let topic = self.topic(topic);
+ let mut newest = topic.newest.subscribe();
+ let mut load = Some(load);
+ loop {
+ if let Some(snapshot) = newest
+ .borrow_and_update()
+ .as_ref()
+ .filter(|snapshot| snapshot.started >= since)
+ {
+ return Some(Arc::clone(snapshot));
+ }
+ if Instant::now() >= deadline {
+ return None;
+ }
+ if load.is_some()
+ && let Ok(running) = Arc::clone(&topic.refresh).try_lock_owned()
+ && let Some(load) = load.take()
+ {
+ let topic = Arc::clone(&topic);
+ let pending = load();
+ tokio::spawn(async move {
+ let started = Instant::now();
+ let probe = pending.await;
+ // Unlock first. A waiter that this wakes and that wants a newer probe must
+ // find the lock free, or it waits out its deadline.
+ drop(running);
+ topic
+ .newest
+ .send_replace(Some(Arc::new(Snapshot { started, probe })));
+ });
+ }
+ timeout_at(deadline, newest.changed()).await.ok()?.ok()?;
+ }
+ }
+
+ /// The newest probe of `topic`, however old. Loads nothing.
+ pub fn newest(&self, topic: &str) -> Option<Arc<Snapshot>> {
+ let topics = self.topics.lock().unwrap_or_else(PoisonError::into_inner);
+ topics.by_name.get(topic)?.newest.borrow().clone()
+ }
+
+ /// Marks a gateway write to `topic` that ended just now. Wakes the Fetches that wait on it.
+ pub fn wrote(&self, topic: &str) {
+ let topics = self.topics.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(topic) = topics.by_name.get(topic) {
+ topic.written.send_replace(Some(Instant::now()));
+ }
+ }
+
+ /// Changes on each [`Self::wrote`] for `topic` after this call, to the time of that write.
+ pub fn writes(&self, topic: &str) -> watch::Receiver<Option<Instant>> {
+ self.topic(topic).written.subscribe()
+ }
+
+ fn topic(&self, name: &str) -> Arc<Topic> {
+ let mut topics = self.topics.lock().unwrap_or_else(PoisonError::into_inner);
+ if let Some(topic) = topics.by_name.get(name) {
+ return Arc::clone(topic);
+ }
+ if topics.by_name.len() >= topics.sweep_at {
+ topics.sweep(Instant::now());
+ }
+ Arc::clone(topics.by_name.entry(name.to_owned()).or_default())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+ use std::time::Duration;
+
+ use tokio::sync::Notify;
+ use tokio::time::sleep;
+
+ use super::*;
+
+ /// A load that counts itself and ends when `release` says so.
+ fn gated(
+ loads: &Arc<AtomicUsize>,
+ release: &Arc<Notify>,
+ ) -> impl FnOnce() -> std::pin::Pin<Box<dyn Future<Output = TopicProbed> + Send>> {
+ let (loads, release) = (Arc::clone(loads), Arc::clone(release));
+ move || {
+ loads.fetch_add(1, Ordering::SeqCst);
+ Box::pin(async move {
+ release.notified().await;
+ Ok(Arc::new(TopicProbe::default()))
+ })
+ }
+ }
+
+ fn later(duration: Duration) -> Instant {
+ Instant::now() + duration
+ }
+
+ /// A load that ends at once.
+ fn done() -> std::future::Ready<TopicProbed> {
+ std::future::ready(Ok(Arc::new(TopicProbe::default())))
+ }
+
+ fn names(board: &ProbeBoard) -> Vec<String> {
+ let mut names: Vec<String> = board
+ .topics
+ .lock()
+ .unwrap()
+ .by_name
+ .keys()
+ .cloned()
+ .collect();
+ names.sort();
+ names
+ }
+
+ #[tokio::test]
+ async fn given_two_fetches_when_both_want_a_new_probe_should_share_one_load() {
+ let board = ProbeBoard::default();
+ let (loads, release) = (Arc::new(AtomicUsize::new(0)), Arc::new(Notify::new()));
+ let (since, deadline) = (Instant::now(), later(Duration::from_secs(5)));
+
+ let (first, second, ()) = tokio::join!(
+ board.probe("a", since, deadline, gated(&loads, &release)),
+ board.probe("a", since, deadline, gated(&loads, &release)),
+ async {
+ sleep(Duration::from_millis(20)).await;
+ release.notify_one();
+ },
+ );
+ assert!(Arc::ptr_eq(&first.unwrap(), &second.unwrap()));
+ assert_eq!(loads.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn given_a_probe_when_a_newer_one_is_wanted_should_load_again() {
+ let board = ProbeBoard::default();
+ let (loads, release) = (Arc::new(AtomicUsize::new(0)), Arc::new(Notify::new()));
+ let deadline = later(Duration::from_secs(5));
+ let probe = |since| {
+ release.notify_one();
+ board.probe("a", since, deadline, gated(&loads, &release))
+ };
+
+ let first = probe(Instant::now()).await.unwrap();
+ let again = probe(first.started).await.unwrap();
+ assert!(Arc::ptr_eq(&first, &again), "new enough");
+ assert_eq!(loads.load(Ordering::SeqCst), 1);
+
+ sleep(Duration::from_millis(1)).await;
+ let newer = probe(Instant::now()).await.unwrap();
+ assert!(newer.started > first.started);
+ assert_eq!(loads.load(Ordering::SeqCst), 2);
+ }
+
+ #[tokio::test]
+ async fn given_a_load_that_hangs_when_the_deadline_passes_should_give_up() {
+ let board = ProbeBoard::default();
+ let deadline = later(Duration::from_millis(20));
+ let hangs = || std::future::pending::<TopicProbed>();
+ assert!(
+ board
+ .probe("a", Instant::now(), deadline, hangs)
+ .await
+ .is_none()
+ );
+ }
+
+ #[tokio::test]
+ async fn given_old_probes_when_the_board_sweeps_should_keep_only_topics_in_use() {
+ let board = ProbeBoard::default();
+ let deadline = later(Duration::from_secs(5));
+ for name in ["a", "b"] {
+ board
+ .probe(name, Instant::now(), deadline, done)
+ .await
+ .unwrap();
+ }
+ let in_use = board.topic("b");
+ board
+ .topics
+ .lock()
+ .unwrap()
+ .sweep(later(PROBE_INTERVAL * 2));
+ assert_eq!(names(&board), vec!["b"]);
+ drop(in_use);
+ }
+
+ #[tokio::test]
+ async fn given_many_new_names_when_their_probes_age_should_forget_them() {
+ let board = ProbeBoard::default();
+ let deadline = later(Duration::from_secs(5));
+ for index in 0..MIN_SWEEP {
+ let name = format!("t{index}");
+ board
+ .probe(&name, Instant::now(), deadline, done)
+ .await
+ .unwrap();
+ }
+ sleep(PROBE_INTERVAL * 2).await;
+ board
+ .probe("new", Instant::now(), deadline, done)
+ .await
+ .unwrap();
+ assert_eq!(names(&board), vec!["new"]);
+ }
+
+ #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+ async fn given_waiters_for_a_newer_probe_when_a_refresh_ends_should_start_the_next_one() {
+ let board = Arc::new(ProbeBoard::default());
+ let (loads, release) = (Arc::new(AtomicUsize::new(0)), Arc::new(Notify::new()));
+ let running = tokio::spawn({
+ let (board, loads, release) =
+ (Arc::clone(&board), Arc::clone(&loads), Arc::clone(&release));
+ let deadline = later(Duration::from_secs(5));
+ async move {
+ let load = gated(&loads, &release);
+ board.probe("a", Instant::now(), deadline, load).await
+ }
+ });
+ while loads.load(Ordering::SeqCst) == 0 {
+ sleep(Duration::from_millis(1)).await;
+ }
+ // The running refresh started before `since`, so its probe is too old for the waiters.
+ sleep(Duration::from_millis(20)).await;
+ let since = Instant::now();
+ let waiters: Vec<_> = (0..4)
+ .map(|_| {
+ let board = Arc::clone(&board);
+ let deadline = later(Duration::from_secs(1));
+ tokio::spawn(async move { board.probe("a", since, deadline, done).await })
+ })
+ .collect();
+ sleep(Duration::from_millis(20)).await;
+ release.notify_one();
+ assert!(running.await.unwrap().is_some());
+ for waiter in waiters {
+ let snapshot = waiter
+ .await
+ .unwrap()
+ .expect("a newer probe before the deadline");
+ assert!(snapshot.started >= since);
+ }
+ }
+
+ #[tokio::test]
+ async fn given_a_probe_when_asked_for_the_newest_should_return_it_without_a_load() {
+ let board = ProbeBoard::default();
+ assert!(board.newest("a").is_none(), "nothing loads here");
+ assert!(names(&board).is_empty(), "and no topic joins the board");
+ let since = Instant::now();
+ board
+ .probe("a", since, later(Duration::from_secs(1)), done)
+ .await
+ .expect("the load ends at once");
+ let newest = board.newest("a").expect("the probe is on the board");
+ assert!(newest.started >= since);
+ }
+}
diff --git a/gateways/kafka/src/records.rs b/gateways/kafka/src/records.rs
index 53640b755..ae3a1f60f 100644
--- a/gateways/kafka/src/records.rs
+++ b/gateways/kafka/src/records.rs
@@ -23,7 +23,7 @@
use std::cell::Cell;
use std::collections::BTreeMap;
use std::io::{self, Write};
-use std::sync::OnceLock;
+use std::sync::{LazyLock, OnceLock};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use iggy::prelude::{HeaderKey, HeaderValue, IggyError, IggyMessage, MAX_PAYLOAD_SIZE};
@@ -93,7 +93,7 @@ const MAX_VARINT_BYTES: usize = 5;
const MAX_VARLONG_BYTES: usize = 10;
/// Base offset, batch length, leader epoch, magic, CRC, attributes, last offset delta, first and
/// max timestamp, producer id, producer epoch, base sequence and record count.
-const BATCH_HEADER_BYTES: usize = 61;
+pub const BATCH_HEADER_BYTES: usize = 61;
/// Widest v2 record framing: five varints at five bytes each, an attributes byte, and the header
/// count varint, before the key, the value and the header bytes.
const RECORD_FRAMING_BYTES: usize = 31;
@@ -259,16 +259,27 @@ struct Stamp {
/// enforces the namespace, so that is a claim the message makes and not one the server keeps: an
/// Iggy writer can set `kafka.v` to a version this build does not implement, and every read of
/// that message then fails. Fetch cannot serve a record it cannot decode and a Kafka consumer
-/// cannot step over one, so the handler in #3536 owns the skip-or-quarantine policy for a message
-/// that fails here. `BRIDGE_MAPPING.md` records that as the open end of the provenance design.
+/// cannot step over one, so Fetch answers -1 at that offset and the consumer stops there.
+/// `BRIDGE_MAPPING.md` records that as the open end of the provenance design.
///
/// # Errors
///
/// Returns an error when the stored user headers do not parse, when `kafka.v` names a mapping
/// version this build does not implement, or when a marker or an envelope is malformed.
pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> {
+ // The common record: a value alone, which `plain_message` wrote. Skips the header map.
+ if message.user_headers.as_deref() == Some(version_only()) {
+ let value = Some(message.payload.clone());
+ return Ok(record(
+ None,
+ value,
+ IndexMap::new(),
+ offset,
+ timestamp_out(message),
+ ));
+ }
let stored = user_headers(message)?;
- let Some(version) = stored.get(&header_key(VERSION_HEADER)) else {
+ let Some(version) = stored.get(&VERSION_KEY) else {
let (key, value, headers) = foreign_fields(message, &stored);
return Ok(record(key, value, headers, offset, timestamp_out(message)));
};
@@ -277,11 +288,11 @@ pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> {
return Err(RecordCodecError::MappingVersion(version));
}
- let (key, value, headers) = match stored.get(&header_key(ENVELOPE_HEADER)) {
+ let (key, value, headers) = match stored.get(&ENVELOPE_KEY) {
Some(envelope) => decode_envelope(envelope.as_bytes(), &message.payload)?,
None => gateway_fields(message, &stored)?,
};
- let timestamp = match stored.get(&header_key(TIMESTAMP_HEADER)) {
+ let timestamp = match stored.get(&TIMESTAMP_KEY) {
None => timestamp_out(message),
Some(header) => match header.as_int64() {
Ok(NO_TIMESTAMP) => server_timestamp(message),
@@ -364,17 +375,7 @@ fn plain_value(record: &Record, stamp: Stamp) -> Option<&Bytes> {
///
/// Static bytes: a clone touches no refcount shared across workers.
fn plain_message(value: Bytes, origin: u64) -> Result<IggyMessage> {
- static VERSION_ONLY: OnceLock<&'static [u8]> = OnceLock::new();
- let headers = *VERSION_ONLY.get_or_init(|| {
- let block = IggyMessage::builder()
- .payload(Bytes::from_static(PLACEHOLDER))
- .user_headers(gateway_headers(None))
- .build()
- .ok()
- .and_then(|message| message.user_headers)
- .unwrap_or_else(|| unreachable!("the kafka.v header always fits"));
- Box::leak(block.to_vec().into_boxed_slice())
- });
+ let headers = version_only();
let mut message = IggyMessage::builder().payload(value).build()?;
message.header.user_headers_length =
u32::try_from(headers.len()).unwrap_or_else(|_| unreachable!("the kafka.v header is tiny"));
@@ -540,8 +541,8 @@ fn gateway_fields(
message: &IggyMessage,
stored: &BTreeMap<HeaderKey, HeaderValue>,
) -> Result<RecordFields> {
- let key = stored.get(&header_key(KEY_HEADER)).map(HeaderValue::value);
- let value = match stored.get(&header_key(VALUE_MARKER_HEADER)) {
+ let key = stored.get(&KEY_KEY).map(HeaderValue::value);
+ let value = match stored.get(&VALUE_MARKER_KEY) {
None => Some(message.payload.clone()),
Some(marker) => match marker.as_bytes() {
MARKER_NULL => None,
@@ -658,6 +659,28 @@ fn take_field(buf: &mut Bytes) -> Result<Bytes> {
take(buf, len)
}
+/// The encoded header block that holds `kafka.v` alone.
+fn version_only() -> &'static [u8] {
+ static VERSION_ONLY: OnceLock<&'static [u8]> = OnceLock::new();
+ VERSION_ONLY.get_or_init(|| {
+ let block = IggyMessage::builder()
+ .payload(Bytes::from_static(PLACEHOLDER))
+ .user_headers(gateway_headers(None))
+ .build()
+ .ok()
+ .and_then(|message| message.user_headers)
+ .unwrap_or_else(|| unreachable!("the kafka.v header always fits"));
+ Box::leak(block.to_vec().into_boxed_slice())
+ })
+}
+
+/// Lookup keys for the headers `from_iggy` reads, built once.
+static VERSION_KEY: LazyLock<HeaderKey> = LazyLock::new(|| header_key(VERSION_HEADER));
+static ENVELOPE_KEY: LazyLock<HeaderKey> = LazyLock::new(|| header_key(ENVELOPE_HEADER));
+static TIMESTAMP_KEY: LazyLock<HeaderKey> = LazyLock::new(|| header_key(TIMESTAMP_HEADER));
+static KEY_KEY: LazyLock<HeaderKey> = LazyLock::new(|| header_key(KEY_HEADER));
+static VALUE_MARKER_KEY: LazyLock<HeaderKey> = LazyLock::new(|| header_key(VALUE_MARKER_HEADER));
+
/// Both are infallible for the names and values this module builds: every one is non-empty and
/// within `MAX_FIELD`, which `needs_envelope` guarantees for caller-supplied bytes.
fn header_key(name: &str) -> HeaderKey {
@@ -968,28 +991,28 @@ pub fn encode_batch(records: &mut [Record]) -> Result<Bytes> {
Ok(buf.freeze())
}
+/// An upper bound on the bytes `record` adds to a batch [`encode_batch`] writes.
+///
+/// Fetch sums these against its byte budget before it encodes.
+#[must_use]
+pub fn record_size_bound(record: &Record) -> usize {
+ let field = |field: Option<&Bytes>| field.map_or(0, Bytes::len);
+ RECORD_FRAMING_BYTES
+ + field(record.key.as_ref())
+ + field(record.value.as_ref())
+ + record
+ .headers
+ .iter()
+ .map(|(name, value)| HEADER_FRAMING_BYTES + name.as_str().len() + field(value.as_ref()))
+ .sum::<usize>()
+}
+
/// An upper bound on the encoded batch, so the Fetch path reserves once instead of doubling.
///
/// `kafka_protocol` never reserves: it writes every field with `put_slice` into whatever buffer
/// it is handed. Every length needed here is already in hand.
fn batch_size(records: &[Record]) -> usize {
- let field = |field: Option<&Bytes>| field.map_or(0, Bytes::len);
- BATCH_HEADER_BYTES
- + records
- .iter()
- .map(|record| {
- RECORD_FRAMING_BYTES
- + field(record.key.as_ref())
- + field(record.value.as_ref())
- + record
- .headers
- .iter()
- .map(|(name, value)| {
- HEADER_FRAMING_BYTES + name.as_str().len() + field(value.as_ref())
- })
- .sum::<usize>()
- })
- .sum::<usize>()
+ BATCH_HEADER_BYTES + records.iter().map(record_size_bound).sum::<usize>()
}
/// Decompresses one batch, refusing the write that would pass the budget.
@@ -1265,9 +1288,13 @@ mod tests {
general.header.user_headers_length
);
assert_eq!(plain.header.origin_timestamp, 1_700_000_000_123_000);
- let back = from_iggy(&plain, 0).unwrap();
+ // Fetch reads this block without a header map. It must match the full read.
+ let back = from_iggy(&plain, 7).unwrap();
assert_eq!(back.value.as_deref(), Some(&b"v"[..]));
assert_eq!(back.timestamp, CREATE_TIME);
+ assert_eq!(back.key, None);
+ assert!(back.headers.is_empty());
+ assert_eq!(back.offset, 7);
}
fn envelope_bytes(count: u32, trailing: &[u8]) -> Bytes {
diff --git a/gateways/kafka/tests/fetch_real_bridge_tests.rs b/gateways/kafka/tests/fetch_real_bridge_tests.rs
new file mode 100644
index 000000000..064e9aa11
--- /dev/null
+++ b/gateways/kafka/tests/fetch_real_bridge_tests.rs
@@ -0,0 +1,727 @@
+// 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.
+
+//! Fetch (API key 1) driven through the whole handler against a real `iggy-server`.
+//!
+//! Records go in through the Iggy SDK, mapped by `to_iggy`, and come back as a response that
+//! `kafka_protocol`'s client-side decoder reads, the way a consumer would.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bytes::{Bytes, BytesMut};
+use iggy::prelude::{HeaderKey, HeaderValue, Identifier, IggyMessage, MessageClient, Partitioning};
+use kafka_protocol::indexmap::IndexMap;
+use kafka_protocol::messages::fetch_request::{FetchPartition, FetchTopic};
+use kafka_protocol::messages::fetch_response::PartitionData;
+use kafka_protocol::messages::produce_request::{PartitionProduceData, TopicProduceData};
+use kafka_protocol::messages::{
+ FetchRequest, FetchResponse, ProduceRequest, ProduceResponse, TopicName,
+};
+use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+use kafka_protocol::records::{
+ Compression, NO_PARTITION_LEADER_EPOCH, NO_PRODUCER_EPOCH, NO_PRODUCER_ID, NO_SEQUENCE, Record,
+ RecordBatchDecoder, RecordBatchEncoder, RecordEncodeOptions, TimestampType,
+};
+use serial_test::serial;
+use tokio::time::Instant;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_FETCH, API_KEY_PRODUCE, BrokerAdvertise, GatewayState, handle_request_bounded,
+};
+use iggy_gateway_kafka::records::{MAPPING_VERSION, TimestampWindow, VERSION_HEADER, to_iggy};
+
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+
+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;
+/// The highest version a consumer without topic ids sends.
+const VERSION: i16 = 12;
+/// Java's `max.partition.fetch.bytes` default.
+const PARTITION_MAX_BYTES: i32 = 1024 * 1024;
+/// A Fetch that finds records or an error answers well inside this.
+const PROMPT: Duration = Duration::from_secs(5);
+
+/// 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_UNKNOWN_SERVER_ERROR: i16 = -1;
+const ERROR_NONE: i16 = 0;
+const ERROR_OFFSET_OUT_OF_RANGE: i16 = 1;
+const ERROR_UNKNOWN_TOPIC_OR_PARTITION: i16 = 3;
+const ERROR_NOT_LEADER_OR_FOLLOWER: i16 = 6;
+const ERROR_FETCH_SESSION_ID_NOT_FOUND: i16 = 70;
+const UNKNOWN_OFFSET: i64 = -1;
+
+/// A live gateway with the topic in place, since Fetch 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 fetching from it");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+fn record(key: Option<&[u8]>, value: Option<&[u8]>, headers: &[(&str, Option<&[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: 0,
+ sequence: NO_SEQUENCE,
+ timestamp: CREATE_TIME,
+ key: key.map(Bytes::copy_from_slice),
+ value: value.map(Bytes::copy_from_slice),
+ headers: headers
+ .iter()
+ .map(|(name, value)| {
+ (
+ StrBytes::from_string((*name).to_string()),
+ value.map(Bytes::copy_from_slice),
+ )
+ })
+ .collect::<IndexMap<_, _>>(),
+ }
+}
+
+/// A keyed record holding `value`.
+fn keyed(value: &[u8]) -> Record {
+ record(Some(b"k"), Some(value), &[])
+}
+
+/// Stores `records` in `partition` the way Produce does: `to_iggy`, then one send.
+async fn store(server: &TestServer, partition: u32, records: &[Record]) {
+ let window = TimestampWindow::of(records);
+ let mut messages: Vec<IggyMessage> = records
+ .iter()
+ .map(|record| to_iggy(record, window).expect("the record maps"))
+ .collect();
+ send(server, partition, &mut messages).await;
+}
+
+async fn send(server: &TestServer, partition: u32, messages: &mut [IggyMessage]) {
+ raw_client(server)
+ .await
+ .send_messages(
+ &Identifier::named(STREAM).expect("valid stream name"),
+ &Identifier::named(TOPIC).expect("valid topic name"),
+ &Partitioning::partition_id(partition),
+ messages,
+ )
+ .await
+ .expect("store the messages");
+}
+
+/// A consumer's full request for `topic`, each partition as `(index, fetch_offset)`.
+fn request(topic: &str, partitions: &[(i32, i64)]) -> FetchRequest {
+ FetchRequest::default()
+ .with_max_wait_ms(10_000)
+ .with_min_bytes(1)
+ .with_max_bytes(52_428_800)
+ .with_session_epoch(0)
+ .with_topics(vec![
+ FetchTopic::default()
+ .with_topic(TopicName(StrBytes::from_string(topic.to_string())))
+ .with_partitions(
+ partitions
+ .iter()
+ .map(|&(partition, fetch_offset)| {
+ FetchPartition::default()
+ .with_partition(partition)
+ .with_fetch_offset(fetch_offset)
+ .with_partition_max_bytes(PARTITION_MAX_BYTES)
+ })
+ .collect(),
+ ),
+ ])
+}
+
+/// Sends `request` through the handler at `version`, and decodes the answer as a consumer does.
+async fn fetch_at(state: &GatewayState, version: i16, request: &FetchRequest) -> FetchResponse {
+ let mut body = BytesMut::new();
+ request
+ .encode(&mut body, version)
+ .expect("the request encodes at this version");
+ let mut response = handle_request_bounded(state, API_KEY_FETCH, version, body.freeze())
+ .await
+ .expect_response("Fetch always answers");
+ let decoded = FetchResponse::decode(&mut response, version).expect("a Kafka client decodes it");
+ assert!(response.is_empty(), "no bytes after the response");
+ decoded
+}
+
+/// [`fetch_at`] the highest version, for a request that must not wait.
+async fn fetch(state: &GatewayState, request: &FetchRequest) -> FetchResponse {
+ tokio::time::timeout(PROMPT, fetch_at(state, VERSION, request))
+ .await
+ .expect("records or an error answer at once")
+}
+
+/// The one partition of an answer to a one-partition request.
+fn only(response: &FetchResponse) -> &PartitionData {
+ assert_eq!(response.error_code, ERROR_NONE);
+ assert_eq!(response.session_id, 0, "no session is ever opened");
+ assert_eq!(response.responses.len(), 1);
+ assert_eq!(response.responses[0].partitions.len(), 1);
+ &response.responses[0].partitions[0]
+}
+
+/// Every record a partition answered with.
+fn records(partition: &PartitionData) -> Vec<Record> {
+ let mut bytes = partition.records.clone().unwrap_or_default();
+ RecordBatchDecoder::decode_all(&mut bytes)
+ .expect("only complete batches")
+ .into_iter()
+ .flat_map(|set| set.records)
+ .collect()
+}
+
+fn offsets(partition: &PartitionData) -> Vec<i64> {
+ records(partition)
+ .iter()
+ .map(|record| record.offset)
+ .collect()
+}
+
+fn values(partition: &PartitionData) -> Vec<Option<Bytes>> {
+ records(partition)
+ .into_iter()
+ .map(|record| record.value)
+ .collect()
+}
+
+fn assert_refused(partition: &PartitionData, code: i16) {
+ assert_eq!(partition.error_code, code);
+ assert_eq!(partition.high_watermark, UNKNOWN_OFFSET);
+ assert!(records(partition).is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn given_records_when_fetching_from_zero_should_return_them_in_order() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(
+ &server,
+ 0,
+ &[keyed(b"first"), keyed(b"second"), keyed(b"third")],
+ )
+ .await;
+
+ let response = fetch(&state, &request(TOPIC, &[(0, 0)])).await;
+ let partition = only(&response);
+ assert_eq!(partition.error_code, ERROR_NONE);
+ assert_eq!(partition.high_watermark, 3);
+ assert_eq!(offsets(partition), vec![0, 1, 2]);
+ assert_eq!(
+ values(partition),
+ vec![
+ Some(Bytes::from_static(b"first")),
+ Some(Bytes::from_static(b"second")),
+ Some(Bytes::from_static(b"third")),
+ ]
+ );
+ let first = &records(partition)[0];
+ assert_eq!(first.key.as_deref(), Some(&b"k"[..]));
+ assert_eq!(first.timestamp, CREATE_TIME);
+
+ let from_middle = fetch(&state, &request(TOPIC, &[(0, 1)])).await;
+ assert_eq!(offsets(only(&from_middle)), vec![1, 2]);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_native_iggy_messages_when_fetching_should_return_null_keys() {
+ 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 headers = BTreeMap::new();
+ headers.insert(
+ HeaderKey::try_from("source").expect("valid header name"),
+ HeaderValue::try_from(&b"connector"[..]).expect("valid header value"),
+ );
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload(Bytes::from_static(b"{}"))
+ .user_headers(headers)
+ .build()
+ .expect("valid message"),
+ IggyMessage::from("plain".to_string()),
+ ];
+ send(&server, 0, &mut messages).await;
+
+ let response = fetch(&state, &request(TOPIC, &[(0, 0)])).await;
+ let records = records(only(&response));
+ assert_eq!(records.len(), 2);
+ assert!(records.iter().all(|record| record.key.is_none()));
+ assert_eq!(records[0].value.as_deref(), Some(&b"{}"[..]));
+ assert_eq!(
+ records[0].headers.get(&StrBytes::from_static_str("source")),
+ Some(&Some(Bytes::from_static(b"connector")))
+ );
+ assert_eq!(records[1].value.as_deref(), Some(&b"plain"[..]));
+ assert!(records[1].headers.is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn given_tombstones_and_large_keys_when_fetching_should_round_trip() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let long_key = vec![b'k'; 300];
+ let sent = [
+ record(Some(b"gone"), None, &[]),
+ record(Some(b"k"), Some(b""), &[]),
+ record(Some(&long_key), Some(b"v"), &[]),
+ record(Some(b""), Some(b"v"), &[("flag", None)]),
+ record(None, Some(b"v"), &[("trace", Some(b"abc"))]),
+ ];
+ store(&server, 0, &sent).await;
+
+ let response = fetch(&state, &request(TOPIC, &[(0, 0)])).await;
+ let fetched = records(only(&response));
+ assert_eq!(fetched.len(), sent.len());
+ for (fetched, sent) in fetched.iter().zip(&sent) {
+ assert_eq!(fetched.key, sent.key);
+ assert_eq!(fetched.value, sent.value);
+ assert_eq!(fetched.headers, sent.headers);
+ }
+}
+
+/// Sends `records` as one batch to partition 0 through the Produce handler, and expects them
+/// written.
+async fn produce(state: &GatewayState, records: &[Record]) {
+ const PRODUCE_VERSION: i16 = 9;
+ let mut batch = BytesMut::new();
+ let options = RecordEncodeOptions {
+ version: 2,
+ compression: Compression::None,
+ };
+ RecordBatchEncoder::encode(&mut batch, records.iter(), &options).expect("the batch encodes");
+ let request = ProduceRequest::default()
+ .with_acks(-1)
+ .with_timeout_ms(5_000)
+ .with_topic_data(vec![
+ TopicProduceData::default()
+ .with_name(TopicName(StrBytes::from_static_str(TOPIC)))
+ .with_partition_data(vec![
+ PartitionProduceData::default()
+ .with_index(0)
+ .with_records(Some(batch.freeze())),
+ ]),
+ ]);
+ let mut body = BytesMut::new();
+ request
+ .encode(&mut body, PRODUCE_VERSION)
+ .expect("the request encodes");
+ let mut response =
+ handle_request_bounded(state, API_KEY_PRODUCE, PRODUCE_VERSION, body.freeze())
+ .await
+ .expect_response("acks=-1 answers");
+ let decoded =
+ ProduceResponse::decode(&mut response, PRODUCE_VERSION).expect("a producer decodes it");
+ assert_eq!(
+ decoded.responses[0].partition_responses[0].error_code,
+ ERROR_NONE
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_records_written_through_produce_when_fetching_should_return_them_unchanged() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let stamped = |mut record: Record, timestamp| {
+ record.timestamp = timestamp;
+ record
+ };
+ // The epoch sits far outside the window of the rest, so Produce keeps it in `kafka.ts`.
+ let sent = [
+ record(Some(b"k"), Some(b"v"), &[("trace", Some(b"abc"))]),
+ record(None, Some(b"plain"), &[]),
+ record(Some(b"gone"), None, &[]),
+ record(Some(b""), Some(b""), &[("flag", None)]),
+ stamped(record(None, Some(b"old"), &[]), 0),
+ ];
+ produce(&state, &sent).await;
+
+ let response = fetch(&state, &request(TOPIC, &[(0, 0)])).await;
+ let partition = only(&response);
+ let fetched = records(partition);
+ assert_eq!(partition.high_watermark, 5);
+ assert_eq!(fetched.len(), sent.len());
+ for (offset, (fetched, sent)) in fetched.iter().zip(&sent).enumerate() {
+ assert_eq!(fetched.offset, i64::try_from(offset).unwrap());
+ assert_eq!(fetched.key, sent.key);
+ assert_eq!(fetched.value, sent.value);
+ assert_eq!(fetched.headers, sent.headers);
+ assert_eq!(fetched.timestamp, sent.timestamp);
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn given_offset_at_high_watermark_when_fetching_should_wait_and_return_empty() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(&server, 0, &[keyed(b"a"), keyed(b"b")]).await;
+ let max_wait = Duration::from_millis(300);
+
+ let started = Instant::now();
+ let caught_up = request(TOPIC, &[(0, 2)]).with_max_wait_ms(300);
+ let response = fetch_at(&state, VERSION, &caught_up).await;
+ let waited = started.elapsed();
+
+ assert!(waited >= max_wait, "answered after {waited:?}");
+ assert!(waited < PROMPT, "answered after {waited:?}");
+ let partition = only(&response);
+ assert_eq!(partition.error_code, ERROR_NONE);
+ assert_eq!(partition.high_watermark, 2);
+ assert!(records(partition).is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_write_during_the_wait_when_fetching_should_return_early() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let write_after = Duration::from_millis(300);
+
+ let started = Instant::now();
+ let empty = request(TOPIC, &[(0, 0)]).with_max_wait_ms(10_000);
+ let (response, ()) = tokio::join!(fetch_at(&state, VERSION, &empty), async {
+ tokio::time::sleep(write_after).await;
+ store(&server, 0, &[keyed(b"late")]).await;
+ });
+ let waited = started.elapsed();
+
+ assert!(waited >= write_after, "answered after {waited:?}");
+ assert!(waited < PROMPT, "the record ends the wait, not max_wait_ms");
+ let partition = only(&response);
+ assert_eq!(partition.high_watermark, 1);
+ assert_eq!(values(partition), vec![Some(Bytes::from_static(b"late"))]);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_message_that_does_not_map_when_fetching_should_answer_minus_one_after_the_wait() {
+ 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 headers = BTreeMap::new();
+ headers.insert(
+ HeaderKey::try_from(VERSION_HEADER).expect("valid header name"),
+ HeaderValue::try_from(&[MAPPING_VERSION + 1][..]).expect("valid header value"),
+ );
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload(Bytes::from_static(b"v"))
+ .user_headers(headers)
+ .build()
+ .expect("valid message"),
+ ];
+ send(&server, 0, &mut messages).await;
+ let max_wait = Duration::from_millis(300);
+
+ let started = Instant::now();
+ let stuck = request(TOPIC, &[(0, 0)]).with_max_wait_ms(300);
+ let response = fetch_at(&state, VERSION, &stuck).await;
+ let waited = started.elapsed();
+
+ assert!(
+ waited >= max_wait,
+ "a consumer that retries at once must not spin: {waited:?}"
+ );
+ assert!(waited < PROMPT, "answered after {waited:?}");
+ assert_refused(only(&response), ERROR_UNKNOWN_SERVER_ERROR);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_zero_max_wait_when_fetching_should_answer_at_once() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ for no_wait in [
+ request(TOPIC, &[(0, 0)]).with_max_wait_ms(0),
+ // Kafka has zero bytes in hand, and that meets zero.
+ request(TOPIC, &[(0, 0)]).with_min_bytes(0),
+ ] {
+ let response = fetch(&state, &no_wait).await;
+ let partition = only(&response);
+ assert_eq!(partition.error_code, ERROR_NONE);
+ assert_eq!(partition.high_watermark, 0);
+ assert!(records(partition).is_empty());
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn given_offset_past_high_watermark_when_fetching_should_return_offset_out_of_range() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(&server, 0, &[keyed(b"a"), keyed(b"b")]).await;
+
+ let past = fetch(&state, &request(TOPIC, &[(0, 3)])).await;
+ assert_refused(only(&past), ERROR_OFFSET_OUT_OF_RANGE);
+
+ let negative = fetch(&state, &request(TOPIC, &[(0, -1)])).await;
+ assert_refused(only(&negative), ERROR_OFFSET_OUT_OF_RANGE);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_an_empty_partition_when_fetching_past_it_should_answer_6_after_the_wait() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let max_wait = Duration::from_millis(300);
+
+ // Iggy reports (0, 0) for a partition it loads or fences too. A 1 would reset the consumer.
+ let started = Instant::now();
+ let past = request(TOPIC, &[(0, 5)]).with_max_wait_ms(300);
+ let response = fetch_at(&state, VERSION, &past).await;
+ let waited = started.elapsed();
+ assert!(
+ waited >= max_wait,
+ "a consumer that fetches again at once after a 6 must not spin: {waited:?}"
+ );
+ assert!(waited < PROMPT, "answered after {waited:?}");
+ assert_refused(only(&response), ERROR_NOT_LEADER_OR_FOLLOWER);
+
+ let at_start = request(TOPIC, &[(0, 0)]).with_max_wait_ms(0);
+ let response = fetch(&state, &at_start).await;
+ let partition = only(&response);
+ assert_eq!(partition.error_code, ERROR_NONE, "caught up, not refused");
+ assert!(records(partition).is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn given_unknown_topic_or_partition_when_fetching_should_return_3() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+
+ for (topic, partition) in [
+ ("missing", 0),
+ (TOPIC, 5),
+ (TOPIC, -1),
+ // No topic has a name Kafka refuses, and 17 would make the Java consumer throw.
+ ("bad name", 0),
+ ] {
+ let response = fetch(&state, &request(topic, &[(partition, 0)])).await;
+ assert_refused(only(&response), ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn given_a_small_partition_max_bytes_when_fetching_should_still_return_one_record() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let big = [b'x'; 2048];
+ store(&server, 0, &[keyed(&big), keyed(&big), keyed(&big)]).await;
+
+ let mut small_partition = request(TOPIC, &[(0, 0)]);
+ small_partition.topics[0].partitions[0].partition_max_bytes = 10;
+ let response = fetch(&state, &small_partition).await;
+ let partition = only(&response);
+ assert_eq!(offsets(partition), vec![0], "one record, past the limit");
+ assert_eq!(partition.high_watermark, 3);
+
+ let small_response = request(TOPIC, &[(0, 0)]).with_max_bytes(10);
+ let response = fetch(&state, &small_response).await;
+ assert_eq!(offsets(only(&response)), vec![0]);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_small_then_large_messages_when_fetching_should_page_through_them_in_order() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ let small = keyed(&[b's'; 100]);
+ let large = keyed(&vec![b'l'; 256 * 1024]);
+ store(&server, 0, &vec![small; 100]).await;
+ for _ in 0..2 {
+ store(&server, 0, &vec![large.clone(); 4]).await;
+ }
+ let budget = usize::try_from(PARTITION_MAX_BYTES).expect("positive");
+
+ let from_start = fetch(&state, &request(TOPIC, &[(0, 0)])).await;
+ let partition = only(&from_start);
+ assert_eq!(partition.high_watermark, 108);
+ assert_eq!(
+ offsets(partition),
+ (0..103).collect::<Vec<i64>>(),
+ "every small one, then the large ones that fit, in order"
+ );
+ assert!(partition.records.as_ref().expect("records").len() <= budget);
+
+ let from_large = fetch(&state, &request(TOPIC, &[(0, 100)])).await;
+ let partition = only(&from_large);
+ assert_eq!(offsets(partition), vec![100, 101, 102]);
+ assert!(partition.records.as_ref().expect("records").len() <= budget);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_max_bytes_spent_when_fetching_should_answer_remaining_partitions_empty() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 2).await;
+ let value = [b'x'; 1024];
+ for partition in 0..2 {
+ store(
+ &server,
+ partition,
+ &[keyed(&value), keyed(&value), keyed(&value)],
+ )
+ .await;
+ }
+
+ let response = fetch(&state, &request(TOPIC, &[(0, 0), (1, 0)]).with_max_bytes(1)).await;
+ let partitions = &response.responses[0].partitions;
+ assert_eq!(partitions.len(), 2, "every requested partition answers");
+ assert_eq!(
+ offsets(&partitions[0]),
+ vec![0],
+ "the first gets one record"
+ );
+ assert_eq!(partitions[1].error_code, ERROR_NONE);
+ assert!(records(&partitions[1]).is_empty(), "nothing left to spend");
+ assert_eq!(
+ partitions[1].high_watermark, 3,
+ "the probe's high watermark"
+ );
+}
+
+#[tokio::test]
+#[serial]
+async fn given_an_incremental_session_when_fetching_should_return_70() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(&server, 0, &[keyed(b"a")]).await;
+
+ let incremental = request(TOPIC, &[(0, 0)])
+ .with_session_id(9)
+ .with_session_epoch(1);
+ for version in [7, VERSION] {
+ let response = fetch_at(&state, version, &incremental).await;
+ assert_eq!(response.error_code, ERROR_FETCH_SESSION_ID_NOT_FOUND);
+ assert_eq!(response.session_id, 0);
+ assert!(response.responses.is_empty(), "as Kafka answers");
+ }
+
+ // A full request that names a session closes it, so it reads as usual.
+ let full = request(TOPIC, &[(0, 0)]).with_session_id(9);
+ let response = fetch(&state, &full).await;
+ assert_eq!(offsets(only(&response)), vec![0]);
+}
+
+#[tokio::test]
+#[serial]
+async fn given_every_supported_version_when_fetching_should_decode() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(
+ &server,
+ 0,
+ &[
+ keyed(b"a"),
+ record(Some(b"k"), Some(b"b"), &[("trace", Some(b"abc"))]),
+ ],
+ )
+ .await;
+
+ for version in 4..=12 {
+ let response = tokio::time::timeout(
+ PROMPT,
+ fetch_at(&state, version, &request(TOPIC, &[(0, 0)])),
+ )
+ .await
+ .expect("records answer at once");
+ let partition = only(&response);
+ assert_eq!(partition.error_code, ERROR_NONE, "v{version}");
+ assert_eq!(partition.high_watermark, 2, "v{version}");
+ assert_eq!(partition.last_stable_offset, 2, "v{version}");
+ assert_eq!(partition.log_start_offset, UNKNOWN_OFFSET, "v{version}");
+ assert_eq!(offsets(partition), vec![0, 1], "v{version}");
+ assert_eq!(
+ values(partition),
+ vec![
+ Some(Bytes::from_static(b"a")),
+ Some(Bytes::from_static(b"b"))
+ ],
+ "v{version}"
+ );
+ }
+}
+
+#[tokio::test]
+#[serial]
+async fn given_read_committed_when_fetching_should_set_last_stable_offset_to_high_watermark() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = gateway_with_topic(&server, 1).await;
+ store(&server, 0, &[keyed(b"a"), keyed(b"b")]).await;
+
+ let read_committed = request(TOPIC, &[(0, 0)]).with_isolation_level(1);
+ let response = fetch(&state, &read_committed).await;
+ let partition = only(&response);
+ assert_eq!(partition.high_watermark, 2);
+ assert_eq!(
+ partition.last_stable_offset, 2,
+ "no transactions, so nothing is pending"
+ );
+ assert_eq!(partition.aborted_transactions, None);
+ assert_eq!(offsets(partition), vec![0, 1]);
+}
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index 641c86706..e283f050f 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -27,18 +27,29 @@
//! with a real `IggyAuthenticator`, and a client from a container. They automate categories S and T
//! of `docs/MANUAL_TESTING.md`.
//!
+//! The Fetch tests swap the authenticator for a bridge: kcat writes records through Produce, and
+//! kcat and the Java consumer read them back through Fetch.
+//!
//! Prerequisites are Docker and an already-built `iggy-server`. Missing either skips, the way the
//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns a skip into a failure
//! so a broken CI step cannot leave these silently green.
+use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};
-use iggy_gateway_kafka::GatewayConfig;
+use secrecy::SecretString;
+use tokio::sync::broadcast;
+
use iggy_gateway_kafka::auth::IggyAuthenticator;
+use iggy_gateway_kafka::bridge::{
+ DEFAULT_MAX_MESSAGE_SIZE, IggyBridge, IggyBridgeConfig, TopicMapping,
+};
+use iggy_gateway_kafka::server::bind_listener;
+use iggy_gateway_kafka::{GatewayConfig, KafkaGateway};
#[path = "common/server.rs"]
mod server;
@@ -51,6 +62,10 @@ const ROOT_USER: &str = "iggy";
const ROOT_PASSWORD: &str = "iggy";
/// Password for every non-root principal these tests create.
const USER_PASSWORD: &str = "s3cretpass";
+/// The topic the bridged tests write to and read from. It has one partition.
+const TOPIC: &str = "orders";
+/// What the bridged tests write, as `(key, value)`, and read back in this order.
+const RECORDS: [(&str, &str); 3] = [("k1", "alpha"), ("k2", "beta"), ("k3", "gamma")];
/// Budget for `iggy-server` to start listening. Generous: it is a cold process start, and a debug
/// build on a loaded machine is not quick.
@@ -261,6 +276,40 @@ async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
addr
}
+/// Starts the gateway with a bridge to `server` and SASL off. Creates `TOPIC` first, because
+/// neither Produce nor Fetch creates one.
+async fn spawn_bridged_gateway(server: &TestServer) -> SocketAddr {
+ let config = IggyBridgeConfig {
+ address: server.address.clone(),
+ username: ROOT_USER.to_string(),
+ password: SecretString::from(ROOT_PASSWORD.to_string()),
+ topic_mapping: TopicMapping::new("kafka".to_string(), HashMap::new())
+ .expect("the default mapping is valid"),
+ max_message_size: DEFAULT_MAX_MESSAGE_SIZE,
+ };
+ let bridge = IggyBridge::connect(config)
+ .await
+ .expect("the bridge connects to a ready server");
+ bridge
+ .ensure_stream_and_topic(TOPIC, 1)
+ .await
+ .expect("create the topic");
+ let listener = bind_listener("127.0.0.1:0").expect("bind ephemeral port");
+ let addr = listener.local_addr().expect("local addr");
+ let config = GatewayConfig {
+ bind_addr: addr.to_string(),
+ ..GatewayConfig::default()
+ };
+ let gateway = KafkaGateway::new(config).with_bridge(Some(Arc::new(bridge)));
+ let (shutdown, receiver) = broadcast::channel(1);
+ tokio::spawn(async move {
+ let _ = gateway.run(listener, receiver).await;
+ });
+ // Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange.
+ std::mem::forget(shutdown);
+ addr
+}
+
/// Runs a container against the host network and returns its combined output.
///
/// `--network host` is what lets a containerised client reach a gateway bound to the host's
@@ -428,6 +477,20 @@ fn java_client_config(username: &str, password: &str) -> (tempfile::TempDir, Pat
/// Brings up a server and a gateway, or reports why it could not.
async fn stack() -> Option<(TestServer, SocketAddr)> {
+ let server = ready_server()?;
+ let gateway = spawn_gateway(&server.address).await;
+ Some((server, gateway))
+}
+
+/// As [`stack`], with a bridged gateway and SASL off.
+async fn bridged_stack() -> Option<(TestServer, SocketAddr)> {
+ let server = ready_server()?;
+ let gateway = spawn_bridged_gateway(&server).await;
+ Some((server, gateway))
+}
+
+/// Brings up a server once every prerequisite holds, or reports why it could not.
+fn ready_server() -> Option<TestServer> {
if host_unsupported() || docker_missing() {
return None;
}
@@ -440,15 +503,47 @@ async fn stack() -> Option<(TestServer, SocketAddr)> {
));
return None;
}
- let server = match TestServer::spawn() {
- Ok(server) => server,
+ match TestServer::spawn() {
+ Ok(server) => Some(server),
Err(reason) => {
skip(&reason);
- return None;
+ None
}
- };
- let gateway = spawn_gateway(&server.address).await;
- Some((server, gateway))
+ }
+}
+
+/// Writes `RECORDS` to partition 0 of `TOPIC` through the gateway, with kcat's producer and one
+/// header on each record.
+fn produce_with_kcat(gateway: SocketAddr) {
+ let dir = tempfile::tempdir().expect("create records dir");
+ let path = dir.path().join("records.txt");
+ let lines = RECORDS
+ .map(|(key, value)| format!("{key}:{value}\n"))
+ .join("");
+ std::fs::write(&path, lines).expect("write records");
+ run_client(
+ KCAT_IMAGE,
+ &[
+ "-b",
+ &gateway.to_string(),
+ "-P",
+ "-t",
+ TOPIC,
+ "-p",
+ "0",
+ "-K",
+ ":",
+ "-H",
+ "trace=e2e",
+ "-l",
+ "/tmp/records.txt",
+ ],
+ &[(
+ path.to_str().expect("records path is utf-8"),
+ "/tmp/records.txt",
+ )],
+ )
+ .expect_ran("kcat produce");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -723,3 +818,105 @@ fn list_acls(gateway: SocketAddr, username: &str, password: &str) -> String {
)
.expect_ran(&format!("listing ACLs as {username}"))
}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_records_produced_by_kcat_when_kcat_consumes_should_read_them_back_in_order() {
+ let Some((_server, gateway)) = bridged_stack().await else {
+ return;
+ };
+ produce_with_kcat(gateway);
+ // `-e` stops at the end of the partition, which kcat takes from the high watermark that Fetch
+ // sends. A wrong watermark stops it early or holds it until the container timeout.
+ let output = run_client(
+ KCAT_IMAGE,
+ &[
+ "-b",
+ &gateway.to_string(),
+ "-C",
+ "-t",
+ TOPIC,
+ "-p",
+ "0",
+ "-o",
+ "beginning",
+ "-e",
+ "-f",
+ "record %o %k=%s %h\n",
+ ],
+ &[],
+ )
+ .expect_ran("kcat consume");
+ let read: Vec<&str> = output
+ .lines()
+ .filter(|line| line.starts_with("record "))
+ .collect();
+ let expected: Vec<String> = RECORDS
+ .iter()
+ .enumerate()
+ .map(|(offset, (key, value))| format!("record {offset} {key}={value} trace=e2e"))
+ .collect();
+ assert_eq!(
+ read, expected,
+ "librdkafka must read every record in order, with its key and header, got: {output}"
+ );
+ assert!(
+ output.contains(&format!(
+ "Reached end of topic {TOPIC} [0] at offset {}",
+ RECORDS.len()
+ )),
+ "kcat must stop at the high watermark, got: {output}"
+ );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_records_when_the_java_consumer_reads_the_partition_should_get_them_in_order() {
+ // A second Fetch client with its own code: the Java consumer resolves `earliest` through
+ // ListOffsets, then fetches. `--partition` makes it assign the partition, with no group
+ // offsets to load or commit.
+ let Some((_server, gateway)) = bridged_stack().await else {
+ return;
+ };
+ produce_with_kcat(gateway);
+ let count = RECORDS.len().to_string();
+ let output = run_client(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-console-consumer.sh",
+ "--bootstrap-server",
+ &gateway.to_string(),
+ "--topic",
+ TOPIC,
+ "--partition",
+ "0",
+ "--offset",
+ "earliest",
+ "--max-messages",
+ &count,
+ "--timeout-ms",
+ "20000",
+ "--property",
+ "print.offset=true",
+ "--property",
+ "print.key=true",
+ ],
+ &[],
+ )
+ .expect_ran("java console consumer");
+ let read: Vec<&str> = output
+ .lines()
+ .filter(|line| line.starts_with("Offset:"))
+ .collect();
+ let expected: Vec<String> = RECORDS
+ .iter()
+ .enumerate()
+ .map(|(offset, (key, value))| format!("Offset:{offset}\t{key}\t{value}"))
+ .collect();
+ assert_eq!(
+ read, expected,
+ "the Java consumer must read every record in order, with its key, got: {output}"
+ );
+ assert!(
+ output.contains(&format!("Processed a total of {count} messages")),
+ "the Java consumer must stop after the records it asked for, got: {output}"
+ );
+}