This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch kafka_list_offsets_by_timestamp in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 34e8140bcbe2815ed6dd934cfcb1893e22e4b372 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Oct 14 14:39:48 2026 +0200 temp --- gateways/kafka/README.md | 16 +- gateways/kafka/docs/BRIDGE_MAPPING.md | 57 +- gateways/kafka/docs/MANUAL_TESTING.md | 39 +- gateways/kafka/docs/SCOPE.md | 26 +- gateways/kafka/docs/TEST_SUITE.md | 12 +- gateways/kafka/docs/kafka_api_keys_reference.md | 4 +- gateways/kafka/src/bridge/error.rs | 60 ++ gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 77 ++- gateways/kafka/src/bridge/iggy_bridge/mod.rs | 1 + gateways/kafka/src/bridge/iggy_bridge/offsets.rs | 114 +++- gateways/kafka/src/protocol/api.rs | 8 - .../kafka/src/protocol/handlers/list_offsets.rs | 760 ++++++++++++++++++--- gateways/kafka/src/protocol/handlers/metadata.rs | 2 +- gateways/kafka/src/records.rs | 107 ++- gateways/kafka/tests/api_handler_tests.rs | 2 +- gateways/kafka/tests/common/scope.rs | 2 +- gateways/kafka/tests/common/wire.rs | 2 +- gateways/kafka/tests/golden_wire_fixtures_tests.rs | 4 +- .../kafka/tests/java/OffsetsForTimesProbe.java | 59 ++ gateways/kafka/tests/kafka_client_e2e_tests.rs | 214 +++++- .../kafka/tests/list_offsets_real_bridge_tests.rs | 324 ++++++++- gateways/kafka/tests/version_firewall_tests.rs | 8 +- 22 files changed, 1708 insertions(+), 190 deletions(-) diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 3c83aa194..d7a26a153 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -309,9 +309,9 @@ 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, looks up high watermarks (one or many partitions of -a topic per call) for `ListOffsets`, probes and polls partitions for Fetch, and stores and reads -group offsets for OffsetCommit and OffsetFetch. +streams/topics, provisions them on demand, looks up high watermarks (one or many partitions of a +topic per call) and records by time for `ListOffsets`, probes and polls partitions for Fetch, and +stores and reads group offsets for OffsetCommit and OffsetFetch. 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)), CreateTopics ([#3538](https://github.com/apache/iggy/issues/3538)), DescribeConfigs and AlterConfigs , OffsetCommit and OffsetFetchcall it. @@ -541,7 +541,8 @@ cut before that threshold is reached - worth knowing rather than discovering lat - 4 clients sign in at once, 80 a second at most. More wait, up to their request's deadline. - `IGGY_KAFKA_PRODUCE_SESSIONS=0`: all producers share one client, one request at a time. - Metadata, CreateTopics, DeleteTopics, DescribeConfigs and AlterConfigs share 1 more client, one - Iggy request at a time. Fetch polls use 4 more, one per read slot. Topic probes use 1 more. + Iggy request at a time. Fetch polls use 4 more, one per read slot, and a `ListOffsets` lookup + by time borrows one of those slots for its request. Topic probes use 1 more. Offset calls use 8 more: a commit client and a read client per offset slot. - OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over the offset slots. In a cluster, an offset commit client can also open one connection per node for stores. @@ -589,7 +590,12 @@ costs about 10 `get_topic` calls per second per topic. `ListOffsets` probes on t Fetch folds these into the codes a consumer handles: 7 → 6, 17 → 3, and any other code → -1. OffsetCommit and OffsetFetch fold 6, 7 and `RequestTooOld` into `COORDINATOR_LOAD_IN_PROGRESS` -(14), which the Java client retries in both (`to_offset_error_code`). +(14), which the Java client retries in both (`to_offset_error_code`). ListOffsets answers +`HistoryUnavailable`, `LifecycleBusy` and `ClientShutdown` with 6 and `RequestTooOld` with 7 +instead of -1, because the Java `AdminClient` fails a partition on any non-retriable code +(`to_list_offsets_error_code`). `HistoryUnavailable` refuses a read routed to an older +incarnation of the partition, as after a delete and recreate; the retry probes again and answers 3 +if the topic is gone. ListOffsets reads do not get `RequestTooOld` today, so that arm is defensive. ### Server limits the gateway inherits diff --git a/gateways/kafka/docs/BRIDGE_MAPPING.md b/gateways/kafka/docs/BRIDGE_MAPPING.md index 0e025399d..c31fc4528 100644 --- a/gateways/kafka/docs/BRIDGE_MAPPING.md +++ b/gateways/kafka/docs/BRIDGE_MAPPING.md @@ -300,11 +300,66 @@ Produce takes the base offset from the send confirmation example for a request it classifies as a duplicate, in which case the response carries `-1` rather than a guessed offset; Kafka clients surface that as an unknown offset. -ListOffsets LATEST is the high watermark from `IggyBridge::high_watermarks`. EARLIEST has no +ListOffsets LATEST is the high watermark of the topic probe `IggyBridge::partition_probes` reads +on the bridge's probe client. EARLIEST has no 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. +ListOffsets by timestamp ([#4516](https://github.com/apache/iggy/issues/4516)) polls one record +by time and answers its offset and its timestamp as Fetch reports it: the producer timestamp when +one was set, otherwise the append time. + +| Request | Answer | +| --- | --- | +| A timestamp | The first record Iggy appended at or after it | +| A timestamp past the last record, or any time on a partition that retains no message | No error, offset and timestamp `-1`, as Kafka answers | +| A lookup that polls empty without proof that no record matches (below) | `NOT_LEADER_OR_FOLLOWER` (6), which clients retry | +| `MAX_TIMESTAMP` (`-3`, v7) | The last record | +| A negative time other than the sentinels | The first record, as on a Kafka broker | + +Iggy searches by the time it appended each batch, not by the producer timestamp. A Kafka broker +searches a `CreateTime` topic by producer timestamps, so the two differ when producer clocks +disagree with Iggy's or records arrive late: a record stamped earlier by its producer but +appended after the time is found here and skipped there. With clocks in step the difference is +the delay between send and append. `MAX_TIMESTAMP` is the last record for the same reason, where +Kafka would name the record with the largest producer timestamp. + +The timestamp answered is the producer timestamp when one was set, while the search runs on +append time. So on a `CreateTime` topic the answer can carry a timestamp earlier than the one +asked for, which a Kafka broker never sends. Clients seek by the offset, but +`OffsetAndTimestamp.timestamp()` shows the timestamp to applications as it is. + +Iggy can poll empty while records remain: on a read fault, or before a partition seeds its commit +point. In a cluster the polls can also reach a node behind the one the probe read, which serves a +shorter tail. An empty lookup therefore answers `-1` only when: + +- the probe read a high watermark of 0, so nothing was ever appended; +- a poll of the last record shows it was appended before the time and its offset + 1 reaches the + probed high watermark, so the polled node is not behind the probe; or +- the probe counted no message and the poll of the last record comes back empty too. + +Otherwise it answers 6 and the client asks again, since a `-1` would make +`kafka-consumer-groups --to-datetime` fall back to LATEST and skip the records. + +The server reports a message count of 0 both when retention removed every message and when it has +no stats for the partition, with the high watermark above 0 in both. So a count of 0 alone answers +nothing, and the lookup polls. When both polls come back empty as well, it answers `-1`, as Kafka +does for a partition that retains nothing. A node that has no stats for a partition and also polls +it empty while it holds records answers `-1` too: nothing the probe or the polls return tells the +two apart, and answering 6 instead would never let a reset by time finish on an expired partition. + +A partition Iggy is still loading can read `(messages_count, current_offset) == (0, 0)`, the same +as an empty one, and the probe cannot tell them apart without a server-side signal. A lookup by +time in that window answers `-1`, and a reset by time there moves to LATEST. + +Lookups by time run one partition at a time and reuse one Fetch read slot, however many +partitions the request names. A poll that times out frees its slot as it stops, so for a moment +the request can hold a second one. Each lookup gets a quarter of the request deadline, capped by +the deadline itself. One whose time has already passed answers `REQUEST_TIMED_OUT` (7) without +taking or using a slot, and one that does not finish in time answers 7 too. Clients retry 7. +Repeated entries for one partition and time are looked up once. + ### Header order Kafka carries record headers as an ordered list. Iggy keys them in a `BTreeMap`, so Fetch emits diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index 21c88dd13..d9ee15c87 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -214,7 +214,7 @@ For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-mes | 3 | Metadata | 0 | 9 | −1, 0, 9, 10 | | 0 | Produce | 3 | 9 | 2, 3, 9, 10 | | 1 | Fetch | 4 | 12 | 3, 4, 12, 13 | -| 2 | ListOffsets | 1 | 6 | 0, 1, 6, 7 | +| 2 | ListOffsets | 1 | 7 | 0, 1, 7, 8 | | 19 | CreateTopics | 2 | 5 | 1, 2, 5, 6 | | 20 | DeleteTopics | 1 | 5 | 0, 1, 5, 6 | | 8 | OffsetCommit | 2 | 9 | 1, 2, 9, 10 | @@ -369,6 +369,38 @@ Record wall-clock time from `git clone` to I3 passing - README.md's Quick start target for a new contributor ([#3539](https://github.com/apache/iggy/issues/3539) acceptance criteria). +### Category J — ListOffsets by timestamp with AKHQ + +AKHQ reads a topic's first and last offsets with `ListOffsets` EARLIEST and LATEST, and its +time search with `offsetsForTimes`. It is not in the automated suite, which would need one more +CI image. Start Iggy with `iggy`/`iggy` root credentials and the gateway with +`IGGY_KAFKA_BRIDGE_ENABLED=true` on `127.0.0.1:9093`, create stream `kafka` with topic `orders` +(2 partitions), and send a few messages to each partition at different times. Then run AKHQ on +the host network with this `application.yml`: + +```yaml +micronaut: + server: + port: 18080 +akhq: + connections: + iggy: + properties: + bootstrap.servers: "127.0.0.1:9093" +``` + +```bash +docker run -d --name akhq-iggy --network host \ + -v "$PWD/application.yml:/app/application.yml:ro" tchiotludo/akhq:0.25.1 +``` + +| ID | Test | Steps | Expected | +| ---- | ------ | ------- | ---------- | +| J1 | Topic sizes | `curl -s 'http://127.0.0.1:18080/api/iggy/topic?show=ALL'` | `orders` with `firstOffset` 0 and `lastOffset` equal to each partition's record count, and `size` the total | +| J2 | Records | `curl -s 'http://127.0.0.1:18080/api/iggy/topic/orders/data?sort=OLDEST'` (also `sort=NEWEST`) | Every record, with its partition, offset, timestamp and value | +| J3 | Time search | Add `×tamp=<ISO-8601 time between two sends>` to J2 | Only the records sent after that time | +| J4 | Teardown | `docker rm -f akhq-iggy` | Container removed | + --- ## 4. Validation reference @@ -380,8 +412,8 @@ criteria). | -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. Fetch with a bridge: Iggy unreachable, too slow, loading the partition, or an offset past the end in its first 30 s on the connection | -| 7 | REQUEST_TIMED_OUT | Produce with a bridge: deadline passed, or connection lost mid-send (may be stored) | +| 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, loading the partition, or an offset past the end in its first 30 s on the connection. ListOffsets with a bridge: partition loading, Iggy refused the read for now, or a lookup by time polled empty without proof that no record matches | +| 7 | REQUEST_TIMED_OUT | Produce with a bridge: deadline passed, or connection lost mid-send (may be stored). ListOffsets with a bridge: past the 100-topic cap, the request deadline passed, or a lookup by time ran out of its time | | 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 | | 21 | INVALID_REQUIRED_ACKS | Produce with a bridge: `acks` is not 0, 1 or -1 | @@ -461,6 +493,7 @@ kcat version (if used): ___________ [ ] G1–G11 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3; G11's default `--describe` is expected to fail) [ ] H1–H6 Adversarial input [ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is restart durability) +[ ] J1–J4 AKHQ ListOffsets by timestamp Automated regression: [ ] cargo test -p iggy-gateway-kafka — all passed (see `TEST_SUITE.md` for why this checklist diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index 5a3ef4695..953639ff3 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -53,7 +53,7 @@ it knows the server supports flexible encoding. | 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 | 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 | +| 2 | ListOffsets | 1 | 7 | 1, 2, 3, 4, 5, 6, 7 | With a bridge: LATEST, EARLIEST, a timestamp and, from v7, `MAX_TIMESTAMP` (see `BRIDGE_MAPPING.md`). Without one: stub response | | 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default at every version | | 20 | DeleteTopics | 1 | 5 | 1, 2, 3, 4, 5 | With a bridge: deletes the Iggy topic, never the backing stream (see `BRIDGE_MAPPING.md`). Without one: stub returns `NOT_CONTROLLER` (41). v6 (topic-id-based) not advertised - this bridge has no topic-id concept | | 8 | OffsetCommit | 2 | 9 | 2 … 9 | With a bridge: stores each offset as an Iggy external group offset, and a negative offset deletes it ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). Without one: `COORDINATOR_LOAD_IN_PROGRESS` (14). Flexible encoding at v8+ | @@ -81,7 +81,7 @@ Use this table when configuring clients or generating wire fixtures with `kafka- | --------- | ------ | ---------------------------------- | ---------------------------- | | 0 | Produce | 3–9 | v9 | | 1 | Fetch | 4–12 | v12 | -| 2 | ListOffsets | 1–6 | v6 | +| 2 | ListOffsets | 1–7 | v6 | | 3 | Metadata | 0–9 | v9 | | 8 | OffsetCommit | 2–9 | v8 | | 9 | OffsetFetch | 1–9 | v6 | @@ -336,12 +336,10 @@ below it are still open for the issues that build on top of it. session, so not carried forward as a fix here rather than guessed at. Needs fresh reproduction against a real `kcat` before it's re-closed. - [x] Real ListOffsets ([#3537](https://github.com/apache/iggy/issues/3537)): with - `IGGY_KAFKA_BRIDGE_ENABLED=true`, `LATEST` answers from `IggyBridge::high_watermarks` and - `EARLIEST` answers `0`. Any other requested timestamp (arbitrary-timestamp offset search, - including `offsetsForTimes`/`by_duration` resets) is unsupported - Iggy exposes no - per-message timestamp index - and answers `UNSUPPORTED_FOR_MESSAGE_FORMAT` (43) per - partition rather than a fabricated offset; non-retriable, so a Java client resolves this - immediately instead of spinning until `default.api.timeout.ms`. + `IGGY_KAFKA_BRIDGE_ENABLED=true`, `LATEST` answers from `IggyBridge::partition_probes` and + `EARLIEST` answers `0`. A timestamp answers the first record Iggy appended at or after it, + and `MAX_TIMESTAMP` (v7) the last record ([#4516](https://github.com/apache/iggy/issues/4516)); + `BRIDGE_MAPPING.md` has the difference from Kafka on `CreateTime` topics. `src/protocol/handlers/list_offsets.rs`, `tests/list_offsets_real_bridge_tests.rs`. With the bridge off, the stub from #3421 answers `NOT_LEADER_OR_FOLLOWER` (6) as before. - `EARLIEST = 0` is real *only* for a partition this bridge has never had retention trim: Iggy @@ -353,16 +351,18 @@ below it are still open for the issues that build on top of it. 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 - entries costs one round trip per entry), and a topic whose every partition asks for an - unsupported timestamp skips the call entirely. A request naming more than 100 distinct topics + one topic probe before any bridge work starts, and every topic gets that probe whatever its + partitions ask for. Repeated (topic, partition, timestamp) entries share one record lookup. + A request naming more than 100 distinct topics resolves the first 100 and answers the rest `REQUEST_TIMED_OUT` (retriable) with no bridge call at all, so a client that retries only its still-erroring topics narrows below the cap on its own. The whole request's aggregate bridge work runs under one 20s wall-clock deadline (`ListOffsets` carries no `timeout_ms` field in any version this gateway supports, so this is a 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`. + batch: a topic already resolved when the deadline arrives keeps its real answer, and the + not-yet-started topics answer `REQUEST_TIMED_OUT`. Each record lookup by time also gets at + most a quarter of that deadline, and answers `REQUEST_TIMED_OUT` past it, so one stalled + partition leaves time for the rest. - [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 diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index 0a62468fd..fae2460f0 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -78,9 +78,9 @@ 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. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. The only suite that can catch a client-compatibility bug, [...] +| [`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. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. Lookups by time resolve through `kafka-get-offsets` (a ti [...] | [`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) | +| [`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, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | 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) | | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) | @@ -95,10 +95,10 @@ is test-only primitive encode/decode scaffolding for hand-building legacy/advers ## Real-client end-to-end suite -`kafka_client_e2e_tests.rs` needs two things the rest of the suite does not: Docker, and an -already-built `iggy-server` in the same target directory. Missing either makes it skip with a -printed reason rather than fail, which is what lets `cargo test -p iggy-gateway-kafka` stay usable -without either. +`kafka_client_e2e_tests.rs` needs three things the rest of the suite does not: Docker, an +already-built `iggy-server` in the same target directory, and a host JDK 17+ (`javac`) for the +consumer `offsetsForTimes` probe. Missing any of them makes it skip with a printed reason rather +than fail, which is what lets `cargo test -p iggy-gateway-kafka` stay usable without them. Client containers run with `--pull never`, so pull the two images first. Without them the suite skips and names the missing one. diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md index 59f4f2f9d..5bad9e9a2 100644 --- a/gateways/kafka/docs/kafka_api_keys_reference.md +++ b/gateways/kafka/docs/kafka_api_keys_reference.md @@ -55,7 +55,7 @@ Key new minimums: | :---: | ---------- | :---------: | :---------: | :-------------: | :--------------: | | 0 | **Produce** | 3 † | 12 | v9 | 🔴 Bridge | | 1 | **Fetch** | 4 | 17 | v12 | 🔴 Bridge | -| 2 | **ListOffsets** | 1 | 9 | v6 | 🟠 Required Stub | +| 2 | **ListOffsets** | 1 | 9 | v6 | 🔴 Bridge | | 3 | **Metadata** | 1 | 12 | v9 | 🔴 Bridge | † See KAFKA-18659 librdkafka workaround above — advertise min=0 in ApiVersions response. @@ -288,7 +288,7 @@ Key new minimums: | ----- | :---: | :---: | :---: | | Produce | v3-v9 | v12 | 3 versions behind | | Fetch | v4-v12 | v17 | 5 versions behind | -| ListOffsets | v1-v6 | v9 | 3 versions behind | +| ListOffsets | v1-v7 | v9 | 2 versions behind | | Metadata | v0-v9 | v12 | 3 versions behind | | ApiVersions | v0-v3 | v4 | 1 version behind | | CreateTopics | v2-v5 | v7 | 2 versions behind | diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs index b7108d135..34c5c6a26 100644 --- a/gateways/kafka/src/bridge/error.rs +++ b/gateways/kafka/src/bridge/error.rs @@ -142,6 +142,25 @@ impl BridgeError { } } + /// [`Self::to_kafka_error_code`] for `ListOffsets`. The Java `AdminClient` fails a partition + /// on any code that is neither 6 nor retriable (`ListOffsetsHandler.handlePartitionError`), so + /// these refusals answer 6 or 7 here, not -1. `HistoryUnavailable` refuses a poll that names + /// an older incarnation of the partition, as after a delete and recreate. The retry probes + /// again and answers 3 if the topic is gone, and clients handle 3 and 6 alike. + /// `RequestTooOld` is not returned to `ListOffsets` reads today, so its arm is defensive. + #[must_use] + pub const fn to_list_offsets_error_code(&self) -> i16 { + match self { + Self::Iggy( + IggyError::HistoryUnavailable + | IggyError::LifecycleBusy + | IggyError::ClientShutdown, + ) => ERROR_NOT_LEADER_OR_FOLLOWER, + Self::Iggy(IggyError::RequestTooOld) => ERROR_REQUEST_TIMED_OUT, + _ => self.to_kafka_error_code(), + } + } + /// True when Iggy rejected the bridge's own login. The Kafka client only sees -1. #[must_use] pub const fn is_bridge_login_rejected(&self) -> bool { @@ -313,6 +332,47 @@ mod tests { assert_eq!(err.to_kafka_error_code(), ERROR_UNKNOWN_SERVER_ERROR); } + #[test] + fn given_a_retriable_iggy_refusal_when_mapped_for_list_offsets_should_answer_retriable() { + let retriable = [ + (IggyError::HistoryUnavailable, ERROR_NOT_LEADER_OR_FOLLOWER), + (IggyError::LifecycleBusy, ERROR_NOT_LEADER_OR_FOLLOWER), + (IggyError::ClientShutdown, ERROR_NOT_LEADER_OR_FOLLOWER), + (IggyError::RequestTooOld, ERROR_REQUEST_TIMED_OUT), + ]; + for (error, code) in retriable { + let err = BridgeError::Iggy(error); + assert_eq!( + err.to_kafka_error_code(), + ERROR_UNKNOWN_SERVER_ERROR, + "{err}" + ); + assert_eq!(err.to_list_offsets_error_code(), code, "{err}"); + } + } + + #[test] + fn given_any_other_failure_when_mapped_for_list_offsets_should_keep_the_shared_code() { + let others = [ + BridgeError::InvalidKafkaTopicName { + kafka_topic: " ".to_string(), + reason: "whitespace".to_string(), + }, + BridgeError::Iggy(IggyError::Unauthorized), + BridgeError::Iggy(IggyError::InvalidCredentials), + BridgeError::Iggy(IggyError::TopicNameNotFound(String::new(), String::new())), + BridgeError::Iggy(IggyError::Disconnected), + BridgeError::Timeout, + ]; + for err in others { + assert_eq!( + err.to_list_offsets_error_code(), + err.to_kafka_error_code(), + "{err}" + ); + } + } + #[test] fn invalid_partition_count_maps_to_invalid_partitions() { let err = BridgeError::InvalidPartitionCount { diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs index fd42f5030..019607ede 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs @@ -31,7 +31,8 @@ use tokio::time::{Instant, timeout_at}; use super::offsets::high_watermark; use super::{ - IggyBridge, REQUEST_TIMEOUT, TopicTarget, connect_client, in_slot, sign_out_and_shut_down, + IggyBridge, REQUEST_TIMEOUT, TopicTarget, connect_client, in_slot, permit_by, + sign_out_and_shut_down, }; use crate::bridge::config::IggyBridgeConfig; use crate::bridge::error::BridgeError; @@ -59,12 +60,10 @@ impl FetchPool { } } - /// A free slot. `None` if none frees by `deadline`. + /// A free slot. `None` if none frees before `deadline`: a poll started then would time out at + /// once and drop the slot's client. 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 permit = permit_by(&self.permits, deadline).await?; let client = self .idle .lock() @@ -334,6 +333,28 @@ impl IggyBridge { offset: u64, count: u32, deadline: Instant, + ) -> Option<(Result<PolledMessages, BridgeError>, FetchSlot)> { + self.poll_by( + slot, + kafka_topic, + partition, + PollingStrategy::offset(offset), + count, + deadline, + ) + .await + } + + /// [`Self::poll`] with any polling strategy, so `ListOffsets` can read by timestamp or the + /// last record through the same slots. + pub(crate) async fn poll_by( + self: &Arc<Self>, + slot: FetchSlot, + kafka_topic: &str, + partition: u32, + strategy: PollingStrategy, + count: u32, + deadline: Instant, ) -> Option<(Result<PolledMessages, BridgeError>, FetchSlot)> { let (bridge, client, topic) = ( Arc::clone(self), @@ -342,7 +363,7 @@ impl IggyBridge { ); let poll = async move { bridge - .poll_with(&client, &topic, partition, offset, count, deadline) + .poll_with(&client, &topic, partition, &strategy, count, deadline) .await }; in_slot(slot, deadline, poll).await @@ -353,12 +374,12 @@ impl IggyBridge { client: &LazyClient, kafka_topic: &str, partition: u32, - offset: u64, + strategy: &PollingStrategy, count: u32, deadline: Instant, ) -> Result<PolledMessages, BridgeError> { let target = self.topic_target(kafka_topic)?; - let (consumer, strategy) = (Consumer::default(), PollingStrategy::offset(offset)); + let consumer = Consumer::default(); let poll = async { client .connected(&self.config) @@ -368,7 +389,7 @@ impl IggyBridge { &target.topic_id, Some(partition), &consumer, - &strategy, + strategy, count, false, ) @@ -381,6 +402,27 @@ impl IggyBridge { } } +#[cfg(test)] +impl IggyBridge { + /// Gives every Fetch read slot `client`, so a test needs no server. + pub(crate) async fn hold_fetch_clients(&self, client: &Arc<IggyClient>) { + for slot_client in &self.fetch_pool.all { + slot_client.hold(Arc::clone(client)).await; + } + } + + /// How many Fetch read slots still hold a client. + pub(crate) async fn held_fetch_clients(&self) -> usize { + let mut held = 0; + for slot_client in &self.fetch_pool.all { + if slot_client.held().await.is_some() { + held += 1; + } + } + held + } +} + #[cfg(test)] mod tests { use std::time::Duration; @@ -468,6 +510,21 @@ mod tests { drop((next, taken)); } + #[tokio::test] + async fn given_a_passed_deadline_when_a_slot_is_free_should_take_none_and_keep_its_client() { + let pool = Arc::new(FetchPool::new()); + let held = unconnected(); + for client in &pool.all { + client.hold(Arc::clone(&held)).await; + } + assert!(pool.take(Instant::now()).await.is_none()); + for client in &pool.all { + assert!(client.held().await.is_some()); + } + let soon = Instant::now() + Duration::from_millis(20); + assert!(pool.take(soon).await.is_some(), "the slot stayed free"); + } + fn partition(current_offset: u64, messages_count: u64, size: u64) -> Partition { Partition { id: 0, diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs index fc068e8ec..e72ff32e1 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs @@ -41,6 +41,7 @@ use fetch::{FetchPool, LazyClient}; pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe}; use group_offsets::OffsetPool; pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls}; +pub(crate) use offsets::{LocatedRecord, RecordLookup}; use produce::ProducePool; pub use produce::{ProduceTurn, Producer}; pub use topics::{KafkaTopicMetadata, StreamTopicCache, TopicCreationOutcome, TopicLoad}; diff --git a/gateways/kafka/src/bridge/iggy_bridge/offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs index fde7579f7..1b429b9c8 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/offsets.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/offsets.rs @@ -17,10 +17,59 @@ //! Offset and high-watermark lookups. -use iggy::prelude::Partition; +use std::sync::Arc; + +use iggy::prelude::{IggyMessage, IggyTimestamp, Partition, PollingStrategy}; +use tokio::time::Instant; use super::IggyBridge; +use super::fetch::{FetchSlot, PartitionProbe}; use crate::bridge::error::BridgeError; +use crate::records::record_timestamp; + +/// Which record a `ListOffsets` lookup by time asks for. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum RecordLookup { + /// The first record appended at or after this time, in Kafka milliseconds. + AppendedAtOrAfter(i64), + /// The last record, the latest Iggy appended. Its reported timestamp is the producer's when + /// one was set, so on a `CreateTime` topic it need not be the largest one. + Last, +} + +/// A record a [`RecordLookup`] found. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LocatedRecord { + pub offset: i64, + /// As Fetch reports it: the producer timestamp when one was set, otherwise the append time. + pub timestamp: i64, + /// When Iggy appended it, in microseconds. What a lookup by time matches against. + pub appended_at: u64, +} + +impl LocatedRecord { + /// Whether Iggy appended this record before `millis`, so a lookup by that time cannot match + /// it. + pub fn appended_before(&self, millis: i64) -> bool { + self.appended_at < kafka_millis_to_micros(millis) + } +} + +impl From<&IggyMessage> for LocatedRecord { + fn from(message: &IggyMessage) -> Self { + let offset = i64::try_from(message.header.offset).unwrap_or(i64::MAX); + let appended_at = message.header.timestamp; + // A message the gateway cannot map still has a server time, and the offset is what the + // client seeks to; Fetch reports the mapping failure when it reaches the record. + let timestamp = record_timestamp(message) + .unwrap_or_else(|_| i64::try_from(appended_at / 1000).unwrap_or(-1)); + Self { + offset, + timestamp, + appended_at, + } + } +} impl IggyBridge { /// The high watermark of each of `partitions`, in the same order, from one `probe` of the @@ -39,11 +88,33 @@ impl IggyBridge { kafka_topic: &str, partitions: &[u32], ) -> Result<Vec<(u32, Result<i64, BridgeError>)>, BridgeError> { + Ok(self + .partition_probes(kafka_topic, partitions) + .await? + .into_iter() + .map(|(partition, probe)| { + let watermark = + probe.map(|probe| i64::try_from(probe.high_watermark).unwrap_or(i64::MAX)); + (partition, watermark) + }) + .collect()) + } + + /// [`Self::high_watermarks`], keeping the whole probe of each partition. + /// + /// # Errors + /// + /// See [`Self::high_watermarks`]. + pub(crate) async fn partition_probes( + &self, + kafka_topic: &str, + partitions: &[u32], + ) -> Result<Vec<(u32, Result<PartitionProbe, BridgeError>)>, BridgeError> { let topic = self.probe(kafka_topic).await?; Ok(partitions .iter() .map(|&partition| { - let watermark = match topic.get(partition) { + let probe = match topic.get(partition) { None => Err(BridgeError::PartitionOutOfRange { topic: kafka_topic.to_string(), partition, @@ -54,13 +125,42 @@ impl IggyBridge { topic: kafka_topic.to_string(), partition, }), - Some(probe) => Ok(i64::try_from(probe.high_watermark).unwrap_or(i64::MAX)), + Some(probe) => Ok(probe), }; - (partition, watermark) + (partition, probe) }) .collect()) } + /// Finds the record `lookup` names in `partition`, with the client of `slot`, and hands the + /// slot back. `Ok(None)` when the poll comes back empty, which Iggy also answers on a read + /// fault or before the partition seeds its commit point, so it proves no miss alone. `None` if + /// no answer comes by `deadline`. + /// + /// Iggy matches a time against the broker append time of each batch, so a record is found by + /// when Iggy stored it, not by the producer timestamp a Kafka broker searches on a + /// `CreateTime` topic. The timestamp reported back is the one Fetch shows. + pub(crate) async fn locate_record( + self: &Arc<Self>, + slot: FetchSlot, + kafka_topic: &str, + partition: u32, + lookup: RecordLookup, + deadline: Instant, + ) -> Option<(Result<Option<LocatedRecord>, BridgeError>, FetchSlot)> { + let strategy = match lookup { + RecordLookup::AppendedAtOrAfter(millis) => { + PollingStrategy::timestamp(IggyTimestamp::from(kafka_millis_to_micros(millis))) + } + RecordLookup::Last => PollingStrategy::last(), + }; + let (polled, slot) = self + .poll_by(slot, kafka_topic, partition, strategy, 1, deadline) + .await?; + let located = polled.map(|polled| polled.messages.first().map(LocatedRecord::from)); + Some((located, slot)) + } + /// [`Self::high_watermarks`] for one partition. /// /// # Errors @@ -89,6 +189,12 @@ impl IggyBridge { } } +/// A time before the epoch matches from the first record, as it does on a Kafka broker, and one +/// past what Iggy's microseconds can hold matches nothing. +fn kafka_millis_to_micros(millis: i64) -> u64 { + u64::try_from(millis).unwrap_or(0).saturating_mul(1000) +} + /// 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. diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs index e6038a06d..13e7498ed 100644 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@ -166,14 +166,6 @@ pub const ERROR_INVALID_REQUEST: i16 = 42; /// be stored. `DescribeConfigs` sends this when a requested key is unknown or a stored expiry /// cannot be shown as `retention.ms`. pub const ERROR_INVALID_CONFIG: i16 = 40; -/// `ListOffsets`' code for a timestamp lookup the broker cannot perform. -/// -/// Real brokers send this for an old-message-format log; this bridge sends it for any timestamp -/// other than the two KIP-79 sentinels, since Iggy has no per-message timestamp index at all. -/// Non-retriable, so a Java client resolves immediately instead of retrying -/// [`ERROR_UNKNOWN_SERVER_ERROR`] until its own `default.api.timeout.ms`. -pub const ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT: i16 = - ResponseError::UnsupportedForMessageFormat.code(); /// `CreateTopics`, `DescribeConfigs`, and `AlterConfigs`: the request addressed more distinct /// topics than this bridge admits in one call. /// diff --git a/gateways/kafka/src/protocol/handlers/list_offsets.rs b/gateways/kafka/src/protocol/handlers/list_offsets.rs index 7c4491d73..58e2027fb 100644 --- a/gateways/kafka/src/protocol/handlers/list_offsets.rs +++ b/gateways/kafka/src/protocol/handlers/list_offsets.rs @@ -18,22 +18,24 @@ //! `ListOffsets` (API key 2). use std::collections::{HashMap, HashSet}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; +use std::time::Duration; use bytes::Bytes; use kafka_protocol::messages::list_offsets_request::{ListOffsetsPartition, ListOffsetsTopic}; use kafka_protocol::messages::list_offsets_response::{ ListOffsetsPartitionResponse, ListOffsetsTopicResponse, }; -use kafka_protocol::messages::{ListOffsetsRequest, ListOffsetsResponse}; +use kafka_protocol::messages::{ListOffsetsRequest, ListOffsetsResponse, TopicName}; use tokio::time::Instant; +use crate::bridge::iggy_bridge::{FetchSlot, LocatedRecord, PartitionProbe, RecordLookup}; use crate::bridge::{BridgeError, IggyBridge}; use crate::error::Result; use crate::protocol::api::{ 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, + ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, REQUEST_DEADLINE, + UNKNOWN_OFFSET, }; use crate::protocol::bounds_guard::validate_list_offsets_shape; use crate::protocol::handlers::fetch::{Spells, sight_loading}; @@ -42,17 +44,16 @@ use crate::protocol::handlers::{decode_guarded, decode_request, encode_message, pub const RANGE: ApiVersionRange = ApiVersionRange { api_key: API_KEY_LIST_OFFSETS, min_version: 1, - max_version: 6, + max_version: 7, }; /// Cap on distinct topics one `ListOffsets` request resolves through the bridge in one pass. /// /// `bounds_guard`'s `MAX_REQUEST_ELEMENTS` (4,096) is a pre-decode `DoS` ceiling, not a usability -/// recommendation: each distinct topic here costs one `high_watermarks` round trip against the -/// single lockstep `IggyClient` every Kafka connection on this gateway shares (`README.md`'s -/// "Concurrency ceiling"). Each call takes its own turn on that shared client and releases it -/// before the next, so a large batch does not hold other connections off for its whole duration - -/// only for whichever single call is in flight at a time. 100 keeps a worst-case batch's aggregate +/// recommendation: each distinct topic here costs one `partition_probes` call on the probe client +/// that the Fetch and `ListOffsets` probes of every Kafka connection share. The calls of one request +/// run one after another, so a large batch shares that client with other connections rather than +/// holding it for its whole duration. 100 keeps a worst-case batch's aggregate /// bridge cost small relative to that shared resource while remaining generous for any real /// consumer's offset lookup. A request naming more than this many distinct topics gets the first /// 100 resolved and the rest answered [`ERROR_REQUEST_TIMED_OUT`] with no bridge call at all - a @@ -64,10 +65,19 @@ const MAX_BRIDGE_BACKED_TOPICS: usize = 100; const LATEST_TIMESTAMP: i64 = -1; /// KIP-79 sentinel: the offset of the first message still retained. const EARLIEST_TIMESTAMP: i64 = -2; +/// KIP-734 sentinel, from v7. Kafka answers the record with the largest timestamp, this gateway +/// the last record appended. +const MAX_TIMESTAMP: i64 = -3; -/// [`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>; +/// Most of the request deadline one record lookup may take, so a partition that stalls leaves +/// time for the rest. +const LOOKUP_BUDGET: Duration = Duration::from_secs(REQUEST_DEADLINE.as_secs() / 4); + +/// [`IggyBridge::partition_probes`]'s return type, spelled once for [`resolve_one_partition`]. +type PartitionProbesResult = core::result::Result< + Vec<(u32, core::result::Result<PartitionProbe, BridgeError>)>, + BridgeError, +>; pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { let decoded = decode_request( @@ -88,7 +98,8 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand // 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, &state.loading, &req.topics, deadline).await; + let resolved = resolve_all_topics(bridge, &state.loading, &req.topics, deadline).await; + let topics = locate_records(bridge, resolved, deadline).await; let resp = ListOffsetsResponse::default().with_topics(topics); respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets") } @@ -102,11 +113,11 @@ fn decode(state: &GatewayState, version: i16, body: Bytes) -> Result<ListOffsets /// 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 { - /// A real `high_watermarks` call was made and returned - or every partition requested was an - /// invalid index and there was nothing to call `high_watermarks` for, which is `Ok(vec![])` + /// A real `partition_probes` call was made and returned - or every partition requested was an + /// invalid index and there was nothing to call `partition_probes` for, which is `Ok(vec![])` /// as far as [`resolve_one_partition`] is concerned (every partition in it fails its own /// `u32::try_from` before ever consulting this). - Watermarks(HighWatermarksResult), + Probed(PartitionProbesResult), /// 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`](crate::protocol::api::ERROR_INVALID_REQUEST) so a client's own @@ -119,12 +130,9 @@ enum TopicLookup { /// `order` preserves first-seen order so the topic cap in [`resolve_topic_lookups`] keeps a /// deterministic prefix of the request rather than an arbitrary hash-order subset. /// -/// Deliberately does *not* skip a topic based on what timestamp its partitions ask for: the only -/// place this bridge checks whether a topic exists at all is the `get_topic` call -/// `high_watermarks` makes internally, so a topic whose partitions all ask an unsupported -/// timestamp still needs that same call to tell a nonexistent topic -/// (`ERROR_UNKNOWN_TOPIC_OR_PARTITION`) apart from an existing one with an unsupported timestamp -/// (`ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT`) - skipping it would report the latter for both. +/// Every topic gets the `partition_probes` call, whatever its partitions ask for: it is what tells a +/// missing topic or partition apart, and what spares a lookup by time on an empty or loading +/// partition. fn group_requested_topics(requested: &[ListOffsetsTopic]) -> (Vec<&str>, HashMap<&str, Vec<u32>>) { let mut order: Vec<&str> = Vec::new(); let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new(); @@ -148,10 +156,10 @@ fn group_requested_topics(requested: &[ListOffsetsTopic]) -> (Vec<&str>, HashMap } /// Resolves one [`TopicLookup`] per name in `order`: [`TopicLookup::NotAttempted`] beyond -/// [`MAX_BRIDGE_BACKED_TOPICS`] or once `deadline` has passed, otherwise a real `high_watermarks` +/// [`MAX_BRIDGE_BACKED_TOPICS`] or once `deadline` has passed, otherwise a real `partition_probes` /// call wrapped in [`tokio::time::timeout_at`] against `deadline` so a topic already resolved /// when time runs out keeps its real answer. A topic with no *valid* partition index at all -/// skips the call - there is nothing `high_watermarks` could tell us that would change any +/// skips the call - there is nothing `partition_probes` could tell us that would change any /// partition's answer, since every one of them already fails its own index check. /// /// A partition that reads as loading counts toward its `loading` spell, so the log shows one that @@ -190,7 +198,7 @@ async fn resolve_topic_lookups<'a>( } let partitions = partitions_by_name[name].as_slice(); if partitions.is_empty() { - lookups.insert(name, TopicLookup::Watermarks(Ok(Vec::new()))); + lookups.insert(name, TopicLookup::Probed(Ok(Vec::new()))); continue; } if deadline_exceeded || Instant::now() >= deadline { @@ -206,25 +214,28 @@ async fn resolve_topic_lookups<'a>( continue; } - let result = - match tokio::time::timeout_at(deadline, bridge.high_watermarks(name, partitions)).await - { - Ok(result) => result, - Err(_elapsed) => { - deadline_exceeded = true; - tracing::warn!( - topic = name, - deadline_secs = REQUEST_DEADLINE.as_secs(), - "ListOffsets bridge call for this topic exceeded the request's aggregate \ + let result = match tokio::time::timeout_at( + deadline, + bridge.partition_probes(name, partitions), + ) + .await + { + Ok(result) => result, + Err(_elapsed) => { + deadline_exceeded = true; + tracing::warn!( + topic = name, + deadline_secs = REQUEST_DEADLINE.as_secs(), + "ListOffsets bridge call for this topic exceeded the request's aggregate \ deadline; answering retriable instead of blocking further" - ); - lookups.insert(name, TopicLookup::NotAttempted); - continue; - } - }; + ); + lookups.insert(name, TopicLookup::NotAttempted); + continue; + } + }; if let Err(call_err) = &result { - let kafka_code = call_err.to_kafka_error_code(); + let kafka_code = call_err.to_list_offsets_error_code(); if kafka_code == ERROR_UNKNOWN_TOPIC_OR_PARTITION { // Client-caused (topic doesn't exist / isn't mapped): expected traffic, not // operator-actionable. @@ -233,27 +244,51 @@ async fn resolve_topic_lookups<'a>( tracing::error!(topic = name, %call_err, "ListOffsets bridge lookup failed"); } } - for (partition, watermark) in result.iter().flatten() { - if matches!(watermark, Err(BridgeError::PartitionLoading { .. })) { + for (partition, probe) in result.iter().flatten() { + if matches!(probe, Err(BridgeError::PartitionLoading { .. })) { sight_loading(loading, name, *partition, Instant::now()); } } - lookups.insert(name, TopicLookup::Watermarks(result)); + lookups.insert(name, TopicLookup::Probed(result)); } lookups } +/// One requested topic entry, its partitions answered or left for a record lookup. +struct ResolvedTopic { + name: TopicName, + partitions: Vec<PartitionAnswer>, +} + +/// One requested partition after [`resolve_one_partition`]. +#[derive(Debug)] +enum PartitionAnswer { + Answered(ListOffsetsPartitionResponse), + /// Its timestamp needs a record lookup, which [`locate_records`] runs. + Pending(PendingLookup), +} + +/// A record lookup still to run, with what the probe saw of its partition. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +struct PendingLookup { + index: i32, + partition: u32, + lookup: RecordLookup, + /// Messages the probe counted. 0 also when the server has no stats for the partition. + retained: u64, + high_watermark: u64, +} + /// Resolves every requested topic entry via [`group_requested_topics`] + /// [`resolve_topic_lookups`], then stamps each requested partition with its topic's -/// [`TopicLookup`] outcome. `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - -/// Iggy exposes no per-message timestamp index - so every other requested timestamp gets -/// [`ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT`] rather than a fabricated offset. +/// [`TopicLookup`] outcome. `EARLIEST` and `LATEST` are answered from the watermark here. Every +/// other timestamp is left [`PartitionAnswer::Pending`] for [`locate_records`]. async fn resolve_all_topics( bridge: &IggyBridge, loading: &Mutex<Spells>, requested: &[ListOffsetsTopic], deadline: Instant, -) -> Vec<ListOffsetsTopicResponse> { +) -> Vec<ResolvedTopic> { let (order, partitions_by_name) = group_requested_topics(requested); let lookups = resolve_topic_lookups(bridge, loading, &order, &partitions_by_name, deadline).await; @@ -271,63 +306,254 @@ async fn resolve_all_topics( .iter() .map(|requested| resolve_one_partition(requested, lookup)) .collect(); - ListOffsetsTopicResponse::default() - .with_name(topic.name.clone()) - .with_partitions(partitions) + ResolvedTopic { + name: topic.name.clone(), + partitions, + } }) .collect() } -/// `lookup` is the whole topic's resolution outcome: an errored [`TopicLookup::Watermarks`] is a +/// `lookup` is the whole topic's resolution outcome: an errored [`TopicLookup::Probed`] is a /// call-level failure (e.g. the mapped stream doesn't exist) applying to every partition alike; /// the inner per-partition `Result` inside its `Ok` is [`BridgeError::PartitionOutOfRange`] for one /// bad index among otherwise resolvable ones. fn resolve_one_partition( requested: &ListOffsetsPartition, lookup: &TopicLookup, -) -> ListOffsetsPartitionResponse { - let Ok(partition_index) = u32::try_from(requested.partition_index) else { - return error_response(requested.partition_index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); +) -> PartitionAnswer { + let index = requested.partition_index; + let Ok(partition_index) = u32::try_from(index) else { + return error_answer(index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); }; let results = match lookup { TopicLookup::NotAttempted => { - return error_response(requested.partition_index, ERROR_REQUEST_TIMED_OUT); + return error_answer(index, ERROR_REQUEST_TIMED_OUT); } - TopicLookup::Watermarks(Err(call_err)) => { - return error_response(requested.partition_index, call_err.to_kafka_error_code()); + TopicLookup::Probed(Err(call_err)) => { + return error_answer(index, call_err.to_list_offsets_error_code()); } - TopicLookup::Watermarks(Ok(results)) => results, + TopicLookup::Probed(Ok(results)) => results, }; // `results` preserves the order of the sorted, deduped partition list `resolve_all_topics` - // passed to `high_watermarks` (`IggyBridge::high_watermarks` maps over its input in place), + // passed to `partition_probes` (`IggyBridge::partition_probes` maps over its input in place), // so a binary search is correct here, not just faster than the linear scan this replaced. - let Ok(found) = results.binary_search_by_key(&partition_index, |(index, _)| *index) else { - return error_response(requested.partition_index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); + let Ok(found) = results.binary_search_by_key(&partition_index, |(id, _)| *id) else { + return error_answer(index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); }; - let (_, watermark) = &results[found]; + let (_, probe) = &results[found]; - let watermark = match watermark { - Err(err) => return error_response(requested.partition_index, err.to_kafka_error_code()), - Ok(watermark) => *watermark, + let probe = match probe { + Err(err) => return error_answer(index, err.to_list_offsets_error_code()), + Ok(probe) => *probe, }; + let watermark = i64::try_from(probe.high_watermark).unwrap_or(i64::MAX); - match requested.timestamp { - LATEST_TIMESTAMP => offset_response(requested.partition_index, watermark), + let answer = match requested.timestamp { + LATEST_TIMESTAMP => offset_response(index, watermark), // 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. 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. - _ => error_response( - requested.partition_index, - ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, - ), + EARLIEST_TIMESTAMP => offset_response(index, 0), + // Nothing was ever appended, so no poll is spent on it. A count of 0 alone is not enough: + // the server reports it without stats too. + _ if probe.high_watermark == 0 => not_found_response(index), + timestamp => { + let lookup = match timestamp { + MAX_TIMESTAMP => RecordLookup::Last, + millis => RecordLookup::AppendedAtOrAfter(millis), + }; + return PartitionAnswer::Pending(PendingLookup { + index, + partition: partition_index, + lookup, + retained: probe.messages_count, + high_watermark: probe.high_watermark, + }); + } + }; + PartitionAnswer::Answered(answer) +} + +fn error_answer(partition: i32, error_code: i16) -> PartitionAnswer { + PartitionAnswer::Answered(error_response(partition, error_code)) +} + +/// Kafka's answer when no record matches a timestamp: no error, and -1 for both fields. +fn not_found_response(partition: i32) -> ListOffsetsPartitionResponse { + ListOffsetsPartitionResponse::default() + .with_partition_index(partition) + .with_timestamp(UNKNOWN_OFFSET) + .with_offset(UNKNOWN_OFFSET) +} + +/// Answers every partition [`resolve_all_topics`] left pending, one record lookup at a time +/// on one Fetch read slot, which each lookup reuses. +/// +/// Looked up one by one rather than at once: a consumer resetting a group by time asks for every +/// partition in one request, and running those in parallel would take every read slot from the +/// consumers that are fetching. A lookup that cannot finish within [`LOOKUP_BUDGET`] or by +/// `deadline` answers [`ERROR_REQUEST_TIMED_OUT`], which the client retries. Repeated entries are +/// looked up once. +async fn locate_records( + bridge: &Arc<IggyBridge>, + topics: Vec<ResolvedTopic>, + deadline: Instant, +) -> Vec<ListOffsetsTopicResponse> { + let mut slot = None; + let mut answered: HashMap<(TopicName, PendingLookup), ListOffsetsPartitionResponse> = + HashMap::new(); + let mut responses = Vec::with_capacity(topics.len()); + for topic in topics { + let mut partitions = Vec::with_capacity(topic.partitions.len()); + for partition in topic.partitions { + let pending = match partition { + PartitionAnswer::Answered(answer) => { + partitions.push(answer); + continue; + } + PartitionAnswer::Pending(pending) => pending, + }; + let key = (topic.name.clone(), pending); + let answer = if let Some(answer) = answered.get(&key) { + answer.clone() + } else { + let answer = locate_one(bridge, &mut slot, &topic.name, pending, deadline).await; + answered.insert(key, answer.clone()); + answer + }; + partitions.push(answer); + } + responses.push( + ListOffsetsTopicResponse::default() + .with_name(topic.name) + .with_partitions(partitions), + ); + } + drop(slot); + responses +} + +/// Answers one `pending` lookup, reusing `slot` or taking a new one into it. +async fn locate_one( + bridge: &Arc<IggyBridge>, + slot: &mut Option<FetchSlot>, + topic: &str, + pending: PendingLookup, + request_deadline: Instant, +) -> ListOffsetsPartitionResponse { + let PendingLookup { + index, + partition, + lookup, + .. + } = pending; + let deadline = lookup_deadline(Instant::now(), request_deadline); + let Some(found) = poll_record(bridge, slot, topic, partition, lookup, deadline).await else { + return error_response(index, ERROR_REQUEST_TIMED_OUT); + }; + let last = match (&found, lookup) { + (Ok(None), RecordLookup::AppendedAtOrAfter(_)) => { + match poll_record(bridge, slot, topic, partition, RecordLookup::Last, deadline).await { + Some(last) => Some(last), + None => return error_response(index, ERROR_REQUEST_TIMED_OUT), + } + } + _ => None, + }; + lookup_answer(topic, pending, found, last.as_ref()) +} + +/// One [`IggyBridge::locate_record`] on `slot`, or on a new slot for the first lookup or after a +/// timed-out poll lost it. +/// `None` if `deadline` has passed, no slot frees, or no answer comes by `deadline`. +async fn poll_record( + bridge: &Arc<IggyBridge>, + slot: &mut Option<FetchSlot>, + topic: &str, + partition: u32, + lookup: RecordLookup, + deadline: Instant, +) -> Option<core::result::Result<Option<LocatedRecord>, BridgeError>> { + // A poll started now would time out at once and drop the slot's client. + if Instant::now() >= deadline { + return None; + } + if slot.is_none() { + // A poll that timed out frees its old slot as it stops, so for a moment the request can + // hold two. + *slot = bridge.fetch_slot(deadline).await; + } + let (located, back) = bridge + .locate_record(slot.take()?, topic, partition, lookup, deadline) + .await?; + *slot = Some(back); + Some(located) +} + +fn lookup_deadline(now: Instant, request_deadline: Instant) -> Instant { + request_deadline.min(now + LOOKUP_BUDGET) +} + +/// The answer to one `pending` lookup, from what it `found` and, after an empty +/// `AppendedAtOrAfter`, the `last` record. +/// +/// Iggy also polls empty on a read fault or before a partition seeds its commit point, and a node +/// that lags the probed one can serve a shorter tail. So "no record" needs the last record appended +/// before the time and reaching the probed high watermark, or, where the probe counted no message, +/// a last poll that is empty too. Anything else answers [`ERROR_NOT_LEADER_OR_FOLLOWER`] to retry. +fn lookup_answer( + topic: &str, + pending: PendingLookup, + found: core::result::Result<Option<LocatedRecord>, BridgeError>, + last: Option<&core::result::Result<Option<LocatedRecord>, BridgeError>>, +) -> ListOffsetsPartitionResponse { + let PendingLookup { + index: partition, + lookup, + retained, + high_watermark, + .. + } = pending; + match found { + Ok(Some(record)) => ListOffsetsPartitionResponse::default() + .with_partition_index(partition) + .with_timestamp(record.timestamp) + .with_offset(record.offset), + Err(error) => { + tracing::warn!(topic, partition, %error, "ListOffsets record lookup by time failed"); + error_response(partition, error.to_list_offsets_error_code()) + } + Ok(None) => { + let proven = match (lookup, last) { + (RecordLookup::Last, _) | (_, Some(Ok(None))) => retained == 0, + (RecordLookup::AppendedAtOrAfter(millis), Some(Ok(Some(last)))) => { + last.appended_before(millis) + && u64::try_from(last.offset) + .is_ok_and(|offset| offset.saturating_add(1) >= high_watermark) + } + _ => false, + }; + if proven { + return not_found_response(partition); + } + tracing::warn!( + topic, + partition, + retained, + high_watermark, + ?lookup, + ?last, + "ListOffsets lookup polled empty without proof that no record matches; \ + answering retriable" + ); + error_response(partition, ERROR_NOT_LEADER_OR_FOLLOWER) + } } } @@ -405,73 +631,394 @@ fn stub_partition_response(partition: i32, error_code: i16) -> ListOffsetsPartit #[cfg(test)] mod tests { + use iggy::prelude::IggyError; + use crate::protocol::api::ERROR_NONE; use super::*; + /// The answer [`resolve_one_partition`] gives without a record lookup. + fn resolved( + requested: &ListOffsetsPartition, + lookup: &TopicLookup, + ) -> ListOffsetsPartitionResponse { + match resolve_one_partition(requested, lookup) { + PartitionAnswer::Answered(answer) => answer, + PartitionAnswer::Pending(pending) => panic!("left for a lookup: {pending:?}"), + } + } + + fn pending(lookup: RecordLookup, retained: u64, high_watermark: u64) -> PendingLookup { + PendingLookup { + index: 3, + partition: 3, + lookup, + retained, + high_watermark, + } + } + fn partition(index: i32, timestamp: i64) -> ListOffsetsPartition { ListOffsetsPartition::default() .with_partition_index(index) .with_timestamp(timestamp) } - fn ok_watermarks(entries: &[(u32, i64)]) -> Vec<(u32, core::result::Result<i64, BridgeError>)> { - entries.iter().map(|&(p, w)| (p, Ok(w))).collect() + /// Partitions that retain every message below their high watermark. + fn ok_watermarks( + entries: &[(u32, u64)], + ) -> Vec<(u32, core::result::Result<PartitionProbe, BridgeError>)> { + entries + .iter() + .map(|&(id, high_watermark)| { + let probe = PartitionProbe { + high_watermark, + average_size: 0, + messages_count: high_watermark, + }; + (id, Ok(probe)) + }) + .collect() } #[test] fn latest_resolves_to_the_watermark() { - let lookup = TopicLookup::Watermarks(Ok(ok_watermarks(&[(0, 42)]))); - let resp = resolve_one_partition(&partition(0, LATEST_TIMESTAMP), &lookup); + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, 42)]))); + let resp = resolved(&partition(0, LATEST_TIMESTAMP), &lookup); assert_eq!(resp.error_code, ERROR_NONE); assert_eq!(resp.offset, 42); } #[test] fn earliest_resolves_to_zero_regardless_of_the_watermark() { - let lookup = TopicLookup::Watermarks(Ok(ok_watermarks(&[(0, 42)]))); - let resp = resolve_one_partition(&partition(0, EARLIEST_TIMESTAMP), &lookup); + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, 42)]))); + let resp = resolved(&partition(0, EARLIEST_TIMESTAMP), &lookup); assert_eq!(resp.error_code, ERROR_NONE); assert_eq!(resp.offset, 0); } #[test] - fn an_arbitrary_timestamp_is_unsupported() { - 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, UNKNOWN_OFFSET); + fn a_timestamp_or_max_timestamp_is_left_for_a_record_lookup() { + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, 42)]))); + let expected = [ + ( + 1_700_000_000_000, + RecordLookup::AppendedAtOrAfter(1_700_000_000_000), + ), + (0, RecordLookup::AppendedAtOrAfter(0)), + (MAX_TIMESTAMP, RecordLookup::Last), + (-7, RecordLookup::AppendedAtOrAfter(-7)), + ]; + for (requested, record_lookup) in expected { + let answer = resolve_one_partition(&partition(0, requested), &lookup); + let PartitionAnswer::Pending(pending) = answer else { + panic!("timestamp {requested} answered without a lookup: {answer:?}"); + }; + let want = PendingLookup { + index: 0, + partition: 0, + lookup: record_lookup, + retained: 42, + high_watermark: 42, + }; + assert_eq!(pending, want, "timestamp {requested}"); + } + } + + /// Retention leaves a count of 0 below a high watermark, and so does a server without stats + /// for the partition, so only a lookup tells whether records remain. + #[test] + fn a_timestamp_where_the_probe_counts_no_message_below_a_high_watermark_is_looked_up() { + let uncounted = PartitionProbe { + high_watermark: 42, + average_size: 0, + messages_count: 0, + }; + let lookup = TopicLookup::Probed(Ok(vec![(0, Ok(uncounted))])); + for requested in [1_700_000_000_000, MAX_TIMESTAMP] { + let answer = resolve_one_partition(&partition(0, requested), &lookup); + assert!( + matches!( + answer, + PartitionAnswer::Pending(PendingLookup { + retained: 0, + high_watermark: 42, + .. + }) + ), + "{requested}: {answer:?}" + ); + } + } + + #[test] + fn an_uncounted_partition_whose_lookups_both_poll_empty_is_not_found() { + // Every message expired: the probe and both polls agree nothing is retained. + let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let resp = lookup_answer("orders", pending(time, 0, 42), Ok(None), Some(&Ok(None))); + assert_eq!(answer(&resp), (ERROR_NONE, -1, -1)); + let resp = lookup_answer("orders", pending(RecordLookup::Last, 0, 42), Ok(None), None); + assert_eq!(answer(&resp), (ERROR_NONE, -1, -1)); + } + + #[test] + fn an_uncounted_partition_answers_the_record_a_lookup_finds() { + let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let found = Ok(Some(record(17, REQUESTED_MILLIS + 5))); + let resp = lookup_answer("orders", pending(time, 0, 42), found, None); + assert_eq!(answer(&resp), (ERROR_NONE, 17, REQUESTED_MILLIS + 5)); + } + + #[test] + fn a_timestamp_on_an_empty_partition_is_not_found_without_a_lookup() { + // Kafka's "no such record": error 0 with -1 offset and timestamp, never an error code. + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, 0)]))); + for requested in [1_700_000_000_000, MAX_TIMESTAMP] { + let resp = resolved(&partition(0, requested), &lookup); + assert_eq!(resp.error_code, ERROR_NONE); + assert_eq!(resp.offset, UNKNOWN_OFFSET); + assert_eq!(resp.timestamp, UNKNOWN_OFFSET); + } + } + + #[test] + fn earliest_and_latest_never_answer_no_error_with_offset_minus_one() { + // The async Java consumer dereferences a null for that entry and throws. + for watermark in [0, 42] { + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, watermark)]))); + for requested in [EARLIEST_TIMESTAMP, LATEST_TIMESTAMP] { + let resp = resolved(&partition(0, requested), &lookup); + assert_eq!(resp.error_code, ERROR_NONE); + assert!(resp.offset >= 0, "{requested} on watermark {watermark}"); + } + } + } + + /// The two answers the Java client turns into "no offset", which the async consumer then + /// dereferences on an EARLIEST or LATEST reset: no error with offset -1, and 43 + /// (`OffsetFetcherUtils.handleListOffsetResponse`). A loading, unknown or timed-out partition + /// answers a code the client retries instead. + #[test] + fn earliest_and_latest_never_answer_what_the_java_client_reads_as_no_offset() { + const UNSUPPORTED_FOR_MESSAGE_FORMAT: i16 = + kafka_protocol::ResponseError::UnsupportedForMessageFormat.code(); + let loading = BridgeError::PartitionLoading { + topic: "orders".to_string(), + partition: 0, + }; + let lookups = [ + TopicLookup::Probed(Ok(ok_watermarks(&[(0, 0)]))), + TopicLookup::Probed(Ok(ok_watermarks(&[(0, 42)]))), + TopicLookup::Probed(Ok(vec![(0, Err(loading))])), + TopicLookup::Probed(Ok(vec![])), + TopicLookup::Probed(Err(BridgeError::Timeout)), + TopicLookup::NotAttempted, + ]; + for lookup in &lookups { + for requested in [EARLIEST_TIMESTAMP, LATEST_TIMESTAMP] { + let resp = resolved(&partition(0, requested), lookup); + assert_ne!(resp.error_code, UNSUPPORTED_FOR_MESSAGE_FORMAT); + assert!( + !(resp.error_code == ERROR_NONE && resp.offset == UNKNOWN_OFFSET), + "{requested} must not answer no error with offset -1" + ); + } + } + } + + #[test] + fn a_retriable_iggy_refusal_answers_retriable_at_every_list_offsets_error_site() { + let refused = || BridgeError::Iggy(IggyError::HistoryUnavailable); + for requested in [EARLIEST_TIMESTAMP, LATEST_TIMESTAMP, REQUESTED_MILLIS] { + let topic = TopicLookup::Probed(Err(refused())); + let partition_level = TopicLookup::Probed(Ok(vec![(0, Err(refused()))])); + for lookup in [&topic, &partition_level] { + let resp = resolved(&partition(0, requested), lookup); + assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER, "{requested}"); + } + } + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let resp = lookup_answer("orders", pending(lookup, 42, 42), Err(refused()), None); + assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER); + } + + const REQUESTED_MILLIS: i64 = 1_700_000_000_000; + + fn record(offset: i64, appended_at_millis: i64) -> LocatedRecord { + LocatedRecord { + offset, + timestamp: appended_at_millis, + appended_at: u64::try_from(appended_at_millis).unwrap() * 1000, + } + } + + fn answer(response: &ListOffsetsPartitionResponse) -> (i16, i64, i64) { + (response.error_code, response.offset, response.timestamp) + } + + #[test] + fn a_found_record_answers_its_offset_and_timestamp() { + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let found = Ok(Some(record(17, REQUESTED_MILLIS + 123))); + let resp = lookup_answer("orders", pending(lookup, 42, 42), found, None); + assert_eq!(answer(&resp), (ERROR_NONE, 17, REQUESTED_MILLIS + 123)); + } + + #[test] + fn a_failed_lookup_answers_its_error() { + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let failed = Err(BridgeError::Timeout); + let resp = lookup_answer("orders", pending(lookup, 42, 42), failed, None); + assert_eq!( + resp.error_code, + BridgeError::Timeout.to_list_offsets_error_code() + ); + } + + #[test] + fn an_empty_poll_is_not_found_when_the_last_record_was_appended_before_the_time() { + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + for high_watermark in [41, 42] { + let last = Some(Ok(Some(record(41, REQUESTED_MILLIS - 1)))); + let resp = lookup_answer( + "orders", + pending(lookup, 42, high_watermark), + Ok(None), + last.as_ref(), + ); + assert_eq!(answer(&resp), (ERROR_NONE, -1, -1), "{high_watermark}"); + } + } + + /// The polled node can lag the probed one and serve a tail that ends before the time. + #[test] + fn an_empty_poll_whose_last_record_falls_short_of_the_high_watermark_answers_retriable() { + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let last = Some(Ok(Some(record(40, REQUESTED_MILLIS - 1)))); + let resp = lookup_answer("orders", pending(lookup, 42, 42), Ok(None), last.as_ref()); + assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER); + } + + #[test] + fn an_empty_poll_while_messages_are_retained_without_proof_answers_retriable() { + let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + let unproven = [ + // The last record matches the time, so the empty poll was a fault. + Some(Ok(Some(record(41, REQUESTED_MILLIS)))), + Some(Ok(Some(record(41, REQUESTED_MILLIS + 1)))), + Some(Ok(None)), + Some(Err(BridgeError::Timeout)), + None, + ]; + for last in unproven { + let shown = format!("{last:?}"); + let resp = lookup_answer("orders", pending(lookup, 42, 42), Ok(None), last.as_ref()); + assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER, "{shown}"); + } + } + + #[test] + fn an_empty_last_poll_while_messages_are_retained_answers_retriable() { + let resp = lookup_answer( + "orders", + pending(RecordLookup::Last, 42, 42), + Ok(None), + None, + ); + assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER); + } + + #[test] + fn each_lookup_gets_its_own_budget_within_the_request_deadline() { + let now = Instant::now(); + assert_eq!( + lookup_deadline(now, now + REQUEST_DEADLINE), + now + LOOKUP_BUDGET + ); + assert!(LOOKUP_BUDGET < REQUEST_DEADLINE); + let near = now + LOOKUP_BUDGET / 2; + assert_eq!(lookup_deadline(now, near), near); + } + + #[tokio::test] + async fn a_lookup_past_its_deadline_answers_retriable_without_using_a_read_slot() { + let bridge = Arc::new(IggyBridge::offline()); + let client = Arc::new( + iggy::prelude::IggyClientBuilder::new() + .with_tcp() + .with_server_address("127.0.0.1:1".to_string()) + .build() + .expect("building a client dials nothing"), + ); + bridge.hold_fetch_clients(&client).await; + let held = bridge.held_fetch_clients().await; + let later = Instant::now() + Duration::from_secs(5); + let mut slot = bridge.fetch_slot(later).await; + assert!(slot.is_some()); + let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS); + for lookup in [time, RecordLookup::Last] { + let passed = Instant::now(); + let resp = locate_one( + &bridge, + &mut slot, + "orders", + pending(lookup, 42, 42), + passed, + ) + .await; + assert_eq!(resp.error_code, ERROR_REQUEST_TIMED_OUT, "{lookup:?}"); + assert!(slot.is_some(), "{lookup:?} kept its slot"); + } + drop(slot); + + let passed = Instant::now(); + let topics = vec![ResolvedTopic { + name: TopicName(kafka_protocol::protocol::StrBytes::from_static_str( + "orders", + )), + partitions: vec![PartitionAnswer::Pending(pending(time, 42, 42))], + }]; + let responses = locate_records(&bridge, topics, passed).await; + assert_eq!( + responses[0].partitions[0].error_code, + ERROR_REQUEST_TIMED_OUT + ); + assert_eq!( + bridge.held_fetch_clients().await, + held, + "every slot kept its client" + ); } #[test] fn a_nonexistent_topic_is_unknown_regardless_of_the_requested_timestamp() { - // Regression for the NoLookupNeeded design this replaced: existence is only ever - // established by the high_watermarks call itself, so a call-level error must win over - // the timestamp branch even when the requested timestamp is unsupported - the call is - // never skipped just because nothing would use its watermark. - let lookup = TopicLookup::Watermarks(Err(BridgeError::Timeout)); - let resp = resolve_one_partition(&partition(0, 1_700_000_000_000), &lookup); - assert_eq!(resp.error_code, BridgeError::Timeout.to_kafka_error_code()); + // Existence is only established by the partition_probes call, so its error wins over any + // record lookup the timestamp would otherwise ask for. + let lookup = TopicLookup::Probed(Err(BridgeError::Timeout)); + let resp = resolved(&partition(0, 1_700_000_000_000), &lookup); + assert_eq!( + resp.error_code, + BridgeError::Timeout.to_list_offsets_error_code() + ); } #[test] fn a_topic_beyond_the_cap_or_past_the_deadline_answers_retriable() { - let resp = - resolve_one_partition(&partition(0, LATEST_TIMESTAMP), &TopicLookup::NotAttempted); + let resp = resolved(&partition(0, LATEST_TIMESTAMP), &TopicLookup::NotAttempted); assert_eq!(resp.error_code, ERROR_REQUEST_TIMED_OUT); assert_eq!(resp.offset, UNKNOWN_OFFSET); } #[test] fn a_negative_partition_index_is_rejected_without_consulting_the_bridge_result() { - let lookup = TopicLookup::Watermarks(Ok(vec![])); - let resp = resolve_one_partition(&partition(-1, LATEST_TIMESTAMP), &lookup); + let lookup = TopicLookup::Probed(Ok(vec![])); + let resp = resolved(&partition(-1, LATEST_TIMESTAMP), &lookup); assert_eq!(resp.error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); } #[test] fn a_partition_specific_error_only_affects_that_partition() { - let lookup = TopicLookup::Watermarks(Ok(vec![( + let lookup = TopicLookup::Probed(Ok(vec![( 0, Err(BridgeError::PartitionOutOfRange { topic: "orders".to_string(), @@ -479,23 +1026,26 @@ mod tests { partitions_count: 0, }), )])); - let resp = resolve_one_partition(&partition(0, LATEST_TIMESTAMP), &lookup); + let resp = resolved(&partition(0, LATEST_TIMESTAMP), &lookup); assert_eq!(resp.error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); } #[test] fn a_call_level_error_applies_regardless_of_the_requested_timestamp() { - let lookup = TopicLookup::Watermarks(Err(BridgeError::Timeout)); - let resp = resolve_one_partition(&partition(0, EARLIEST_TIMESTAMP), &lookup); - assert_eq!(resp.error_code, BridgeError::Timeout.to_kafka_error_code()); + let lookup = TopicLookup::Probed(Err(BridgeError::Timeout)); + let resp = resolved(&partition(0, EARLIEST_TIMESTAMP), &lookup); + assert_eq!( + resp.error_code, + BridgeError::Timeout.to_list_offsets_error_code() + ); } #[test] fn a_partition_index_absent_from_a_sorted_result_is_rejected() { // Exercises the binary search on a multi-entry, sorted-by-index result - a single-entry // vec would pass a linear scan and a broken binary search alike. - let lookup = TopicLookup::Watermarks(Ok(ok_watermarks(&[(0, 10), (2, 30), (5, 60)]))); - let resp = resolve_one_partition(&partition(3, LATEST_TIMESTAMP), &lookup); + let lookup = TopicLookup::Probed(Ok(ok_watermarks(&[(0, 10), (2, 30), (5, 60)]))); + let resp = resolved(&partition(3, LATEST_TIMESTAMP), &lookup); assert_eq!(resp.error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); } } diff --git a/gateways/kafka/src/protocol/handlers/metadata.rs b/gateways/kafka/src/protocol/handlers/metadata.rs index 0b1742cd8..746e5935f 100644 --- a/gateways/kafka/src/protocol/handlers/metadata.rs +++ b/gateways/kafka/src/protocol/handlers/metadata.rs @@ -265,7 +265,7 @@ const RESPONSE_BYTES_TOPIC_OVERHEAD: usize = 13; /// Estimated encoded byte cost of one [`TopicResult`]: the fixed per-topic overhead, the name's /// own bytes (compact string: length plus a short varint prefix, charged at a flat +2), and - /// only for a successful result, since `encode_real_response` sends an empty array otherwise - -/// [`RESPONSE_BYTES_PER_PARTITION`] per partition. +/// [`template_partition_bytes`] per partition. /// /// `bounds_guard` cannot see any of this: it charges the request by name count alone, with no way /// to know a name's length, a topic's per-entry overhead, or its partition count before this diff --git a/gateways/kafka/src/records.rs b/gateways/kafka/src/records.rs index 586ded2cc..08320965d 100644 --- a/gateways/kafka/src/records.rs +++ b/gateways/kafka/src/records.rs @@ -280,30 +280,55 @@ pub fn from_iggy(message: &IggyMessage, offset: i64) -> Result<Record> { )); } let stored = user_headers(message)?; - let Some(version) = stored.get(&VERSION_KEY) else { + let timestamp = stored_timestamp(message, &stored)?; + if !stored.contains_key(&VERSION_KEY) { let (key, value, headers) = foreign_fields(message, &stored); - return Ok(record(key, value, headers, offset, timestamp_out(message))); - }; - if version.as_bytes() != [MAPPING_VERSION] { - let version = version.as_bytes().first().copied().unwrap_or_default(); - return Err(RecordCodecError::MappingVersion(version)); + return Ok(record(key, value, headers, offset, timestamp)); } 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(&TIMESTAMP_KEY) { - None => timestamp_out(message), - Some(header) => match header.as_int64() { - Ok(NO_TIMESTAMP) => server_timestamp(message), - Ok(millis) if millis >= EPOCH_TIMESTAMP => millis, - _ => return Err(RecordCodecError::TimestampHeader(header.value())), - }, - }; Ok(record(key, value, headers, offset, timestamp)) } +/// The timestamp [`from_iggy`] reports for `message`, read from its headers alone. +/// +/// # Errors +/// +/// Fails where `from_iggy` fails before it reaches the key, the value or the envelope: unreadable +/// user headers, an unknown `kafka.v`, or a `kafka.ts` that is not a Kafka timestamp. A malformed +/// envelope or value marker does not fail here. +pub fn record_timestamp(message: &IggyMessage) -> Result<i64> { + if message.user_headers.as_deref() == Some(version_only()) { + return Ok(timestamp_out(message)); + } + stored_timestamp(message, &user_headers(message)?) +} + +/// Checks `kafka.v` and reads `kafka.ts`, which only a gateway-written message has. +fn stored_timestamp( + message: &IggyMessage, + stored: &BTreeMap<HeaderKey, HeaderValue>, +) -> Result<i64> { + let Some(version) = stored.get(&VERSION_KEY) else { + return Ok(timestamp_out(message)); + }; + if version.as_bytes() != [MAPPING_VERSION] { + let version = version.as_bytes().first().copied().unwrap_or_default(); + return Err(RecordCodecError::MappingVersion(version)); + } + stored.get(&TIMESTAMP_KEY).map_or_else( + || Ok(timestamp_out(message)), + |header| match header.as_int64() { + Ok(NO_TIMESTAMP) => Ok(server_timestamp(message)), + Ok(millis) if millis >= EPOCH_TIMESTAMP => Ok(millis), + _ => Err(RecordCodecError::TimestampHeader(header.value())), + }, + ) +} + /// The stored user headers, with an unreadable block told apart from an absent one. /// /// `IggyMessage::user_headers_map` folds a header block it cannot parse into `Ok(None)`, which @@ -1673,6 +1698,60 @@ mod tests { assert_eq!(from_iggy(&message, 0).unwrap().timestamp, EPOCH_TIMESTAMP); } + #[test] + fn given_every_record_shape_when_its_timestamp_is_read_alone_should_match_from_iggy() { + let later = CREATE_TIME + 72 * 60 * 1000; + let clamped = [record_at(CREATE_TIME), record_at(later)]; + let unset_among_real = [record_at(CREATE_TIME), record_at(NO_TIMESTAMP)]; + let epoch_among_real = [record_at(CREATE_TIME), record_at(EPOCH_TIMESTAMP)]; + let enveloped = |timestamp| { + record( + Some(Bytes::new()), + Some(Bytes::from_static(b"v")), + IndexMap::new(), + 0, + timestamp, + ) + }; + let mut unreadable = to_iggy_alone(&record_at(CREATE_TIME)).unwrap(); + unreadable.user_headers = Some(Bytes::from_static(b"not a header block")); + let mut foreign_with_ts = BTreeMap::new(); + foreign_with_ts.insert(header_key(TIMESTAMP_HEADER), HeaderValue::from(CREATE_TIME)); + let foreign_with_ts = IggyMessage::builder() + .payload(Bytes::from_static(b"v")) + .user_headers(foreign_with_ts) + .build() + .unwrap(); + let mut messages = vec![ + to_iggy_alone(&record_with(None, Some(b"v"), &[])).unwrap(), + to_iggy_alone(&record_with(Some(b"k"), Some(b"v"), &[("h", Some(b"x"))])).unwrap(), + to_iggy_alone(&record_at(NO_TIMESTAMP)).unwrap(), + to_iggy_alone(&record_at(EPOCH_TIMESTAMP)).unwrap(), + to_iggy_alone(&enveloped(CREATE_TIME)).unwrap(), + to_iggy_alone(&enveloped(EPOCH_TIMESTAMP)).unwrap(), + to_iggy_alone(&enveloped(NO_TIMESTAMP)).unwrap(), + to_iggy(&clamped[1], TimestampWindow::of(&clamped)).unwrap(), + to_iggy(&unset_among_real[1], TimestampWindow::of(&unset_among_real)).unwrap(), + to_iggy(&epoch_among_real[1], TimestampWindow::of(&epoch_among_real)).unwrap(), + message_with(b"v", &[]), + message_with(b"v", &[("trace", b"abc")]), + foreign_with_ts, + gateway_message(b"v", &[(TIMESTAMP_HEADER, b"epoch")]), + message_with(b"v", &[(VERSION_HEADER, &[MAPPING_VERSION + 1])]), + unreadable, + ]; + for (index, message) in messages.iter_mut().enumerate() { + message.header.timestamp = 5_000_000; + let expected = from_iggy(message, 0).map(|record| record.timestamp); + let actual = record_timestamp(message); + match (&expected, &actual) { + (Ok(expected), Ok(actual)) => assert_eq!(actual, expected, "fixture {index}"), + (Err(_), Err(_)) => {} + _ => panic!("fixture {index}: from_iggy {expected:?}, alone {actual:?}"), + } + } + } + #[test] fn given_no_timestamp_when_read_should_use_the_server_timestamp() { let mut message = to_iggy_alone(&record_at(NO_TIMESTAMP)).unwrap(); diff --git a/gateways/kafka/tests/api_handler_tests.rs b/gateways/kafka/tests/api_handler_tests.rs index eff5ff440..333d283ad 100644 --- a/gateways/kafka/tests/api_handler_tests.rs +++ b/gateways/kafka/tests/api_handler_tests.rs @@ -470,7 +470,7 @@ async fn fetch_decodes_request_with_forgotten_topics_and_rack_id() { #[tokio::test] async fn list_offsets_stub_response_returns_retriable_not_leader() { - for version in 1i16..=6 { + for version in 1i16..=7 { let Some(body) = load_fixture_body_or_skip(2, "ListOffsets", version) else { continue; }; diff --git a/gateways/kafka/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs index b263570cf..69d3d4c70 100644 --- a/gateways/kafka/tests/common/scope.rs +++ b/gateways/kafka/tests/common/scope.rs @@ -28,7 +28,7 @@ use iggy_gateway_kafka::protocol::api::BrokerAdvertise; pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[ (0, "Produce", 3, 9), (1, "Fetch", 4, 12), - (2, "ListOffsets", 1, 6), + (2, "ListOffsets", 1, 7), (3, "Metadata", 0, 9), (8, "OffsetCommit", 2, 9), (9, "OffsetFetch", 1, 9), diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs index 9add5afa3..ef3efb4f8 100644 --- a/gateways/kafka/tests/common/wire.rs +++ b/gateways/kafka/tests/common/wire.rs @@ -134,7 +134,7 @@ pub fn build_metadata_all_topics_flexible(version: i16) -> Bytes { enc.freeze() } -/// Minimal `ListOffsets` request for supported versions (v1–v6). +/// Minimal `ListOffsets` request for supported versions (v1–v7). pub fn build_list_offsets_request(version: i16, topic: &str, partition: i32) -> Bytes { let flexible = version >= 6; let mut enc = Encoder::with_capacity(128); diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs index e5cc16210..46edd3359 100644 --- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs +++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs @@ -48,7 +48,7 @@ async fn golden_apiversions_v3_flexible_response_fixture() { 0x14, // compact array count (19+1) 0x00, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, // key 0: Produce 0-9 (advertised) 0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, 0x00, // key 1: Fetch 4-12 - 0x00, 0x02, 0x00, 0x01, 0x00, 0x06, 0x00, // key 2: ListOffsets 1-6 + 0x00, 0x02, 0x00, 0x01, 0x00, 0x07, 0x00, // key 2: ListOffsets 1-7 0x00, 0x03, 0x00, 0x00, 0x00, 0x09, 0x00, // key 3: Metadata 0-9 0x00, 0x08, 0x00, 0x02, 0x00, 0x09, 0x00, // key 8: OffsetCommit 2-9 0x00, 0x09, 0x00, 0x01, 0x00, 0x09, 0x00, // key 9: OffsetFetch 1-9 @@ -84,7 +84,7 @@ async fn golden_apiversions_v1_response_fixture() { 0x00, 0x00, 0x00, 0x13, // api count = 19 0x00, 0x00, 0x00, 0x00, 0x00, 0x09, // key 0: Produce 0-9 (advertised) 0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, // key 1: Fetch 4-12 - 0x00, 0x02, 0x00, 0x01, 0x00, 0x06, // key 2: ListOffsets 1-6 + 0x00, 0x02, 0x00, 0x01, 0x00, 0x07, // key 2: ListOffsets 1-7 0x00, 0x03, 0x00, 0x00, 0x00, 0x09, // key 3: Metadata 0-9 0x00, 0x08, 0x00, 0x02, 0x00, 0x09, // key 8: OffsetCommit 2-9 0x00, 0x09, 0x00, 0x01, 0x00, 0x09, // key 9: OffsetFetch 1-9 diff --git a/gateways/kafka/tests/java/OffsetsForTimesProbe.java b/gateways/kafka/tests/java/OffsetsForTimesProbe.java new file mode 100644 index 000000000..7c8f39524 --- /dev/null +++ b/gateways/kafka/tests/java/OffsetsForTimesProbe.java @@ -0,0 +1,59 @@ +/* + * 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. + */ + +import java.time.Duration; +import java.util.Map; +import java.util.Properties; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndTimestamp; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; + +/** + * Kafka 3.9 consumer probe for {@code KafkaConsumer.offsetsForTimes}, which plain consumers and + * Kafka Streams seek by time with. Flink, Spark and {@code --by-duration} resets go through + * {@code AdminClient.listOffsets} instead. Prints one {@code time=<ms> offset=<offset|null>} line + * per time. + * + * <p>Arguments: bootstrap server, topic, partition, then one or more times in milliseconds. + */ +public final class OffsetsForTimesProbe { + private OffsetsForTimesProbe() {} + + public static void main(String[] args) { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, args[0]); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); + props.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 10_000); + props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10_000); + TopicPartition partition = new TopicPartition(args[1], Integer.parseInt(args[2])); + try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) { + for (int i = 3; i < args.length; i++) { + long time = Long.parseLong(args[i]); + Map<TopicPartition, OffsetAndTimestamp> found = + consumer.offsetsForTimes(Map.of(partition, time), Duration.ofSeconds(10)); + OffsetAndTimestamp entry = found.get(partition); + System.out.println( + "time=" + time + " offset=" + (entry == null ? "null" : entry.offset())); + } + } + } +} diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs index 1ed5d3054..d34aebb4c 100644 --- a/gateways/kafka/tests/kafka_client_e2e_tests.rs +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@ -31,9 +31,10 @@ //! kcat and the Java consumer read them back through Fetch. The group tests run Java consumers in //! one group across a gateway restart and a member leaving. //! -//! 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. +//! Prerequisites are Docker, an already-built `iggy-server`, and a host JDK (`javac`) for the +//! `offsetsForTimes` probe. Missing any 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::{BTreeMap, HashMap}; use std::io::{BufRead, BufReader, Read}; @@ -44,6 +45,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, mpsc}; use std::time::{Duration, Instant}; +use iggy::prelude::IggyTimestamp; use secrecy::SecretString; use serde_json::Value; use tokio::sync::broadcast; @@ -1400,3 +1402,209 @@ fn records_by_partition(output: &[String]) -> BTreeMap<i64, Vec<i64>> { } read } + +/// Milliseconds since the epoch, the unit Kafka timestamps use. +fn now_millis() -> i64 { + let elapsed = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("the clock is past the epoch"); + i64::try_from(elapsed.as_millis()).expect("now fits an i64") +} + +/// `millis` as the `YYYY-MM-DDTHH:mm:SS.sss` that `--to-datetime` reads, in UTC, which is the +/// client container's zone. +fn utc_datetime(millis: i64) -> String { + let micros = u64::try_from(millis).expect("a time after the epoch") * 1000; + IggyTimestamp::from(micros).to_utc_string("%Y-%m-%dT%H:%M:%S%.3f") +} + +/// Runs `command` under the same GNU `timeout` bound as [`run_client`]. +fn output_bounded(command: &[&str], what: &str) -> std::process::Output { + Command::new("timeout") + .args(["--kill-after=10s", CLIENT_RUN_TIMEOUT]) + .args(command) + .output() + .unwrap_or_else(|error| panic!("run {what}: {error}")) +} + +/// Compiles `tests/java/OffsetsForTimesProbe.java` against the `kafka-clients` jar of the client +/// image, into a directory the client container mounts. `None`, after reporting why, when there +/// is no JDK. +fn compile_offsets_for_times_probe() -> Option<tempfile::TempDir> { + let javac = Command::new("javac") + .arg("-version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()); + if !javac { + skip("javac is unavailable, so the consumer offsetsForTimes probe cannot be built"); + return None; + } + let dir = tempfile::tempdir().expect("create probe dir"); + let jar = dir.path().join("kafka-clients.jar"); + let (_, kafka_version) = KAFKA_IMAGE + .rsplit_once(':') + .expect("the client image is pinned to a version tag"); + let image_jar = format!("/opt/kafka/libs/kafka-clients-{kafka_version}.jar"); + let extracted = output_bounded( + &[ + "docker", + "run", + "--rm", + "--pull", + "never", + "--entrypoint", + "cat", + KAFKA_IMAGE, + &image_jar, + ], + "kafka-clients extraction", + ); + assert!( + extracted.status.success(), + "extracting {image_jar} failed: {}", + String::from_utf8_lossy(&extracted.stderr) + ); + std::fs::write(&jar, extracted.stdout).expect("write kafka-clients jar"); + let source = + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/java/OffsetsForTimesProbe.java"); + let (jar, out, source) = ( + jar.to_str().expect("jar path is utf-8"), + dir.path().to_str().expect("probe dir is utf-8"), + source.to_str().expect("probe source path is utf-8"), + ); + let compiled = output_bounded( + &["javac", "--release", "17", "-cp", jar, "-d", out, source], + "javac", + ); + assert!( + compiled.status.success(), + "javac failed: {}", + String::from_utf8_lossy(&compiled.stderr) + ); + Some(dir) +} + +/// #4516 acceptance: Java's lookups by time resolve against what the gateway stored. Three +/// records, then a time, then a fourth record a second later, so every lookup has one right +/// answer whichever clock the client and Iggy stamp with. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_records_on_both_sides_of_a_time_when_java_looks_up_by_time_should_find_the_later_one() + { + let Some((_server, gateway)) = bridged_stack().await else { + return; + }; + produce_with_kcat(gateway, &RECORDS); + tokio::time::sleep(Duration::from_secs(1)).await; + let between = now_millis(); + tokio::time::sleep(Duration::from_secs(1)).await; + produce_with_kcat(gateway, &[("k4", "delta")]); + + let get_offsets = |time: &str| { + run_client( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-get-offsets.sh", + "--bootstrap-server", + &gateway.to_string(), + "--topic", + TOPIC, + "--time", + time, + ], + &[], + ) + .expect_ran(&format!("kafka-get-offsets --time {time}")) + }; + let at = get_offsets(&between.to_string()); + assert!( + at.lines().any(|line| line.trim() == format!("{TOPIC}:0:3")), + "a lookup by time must find the record after the time, got: {at}" + ); + let start = get_offsets("0"); + assert!( + start + .lines() + .any(|line| line.trim() == format!("{TOPIC}:0:0")), + "time 0 finds where the partition starts, as AKHQ asks, got: {start}" + ); + let max = get_offsets("-3"); + assert!( + max.lines() + .any(|line| line.trim() == format!("{TOPIC}:0:3")), + "max-timestamp names the last record, got: {max}" + ); + let future = now_millis() + 3_600_000; + let after = get_offsets(&future.to_string()); + assert!( + !after + .lines() + .any(|line| line.starts_with(&format!("{TOPIC}:0:"))), + "a time past every record has no offset, which the tool prints as nothing, got: {after}" + ); + + let reset = run_client( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-consumer-groups.sh", + "--bootstrap-server", + &gateway.to_string(), + "--group", + "e2e-reset-by-time", + "--topic", + TOPIC, + "--reset-offsets", + "--to-datetime", + &utc_datetime(between), + "--execute", + ], + &[], + ) + .expect_ran("kafka-consumer-groups --reset-offsets --to-datetime"); + // The tool prints its header and rows without a line break between them, so the row is + // matched as a run of fields anywhere in the output. + let fields: Vec<&str> = reset.split_whitespace().collect(); + let moved = fields + .windows(4) + .any(|row| row == ["e2e-reset-by-time", TOPIC, "0", "3"]); + assert!( + moved, + "the group must move to the first record after the time, got: {reset}" + ); + + // The consumer's own `offsetsForTimes`, which plain consumers and Kafka Streams seek by time + // with, not the Admin call the tools above use. Same request on the wire, different client + // code. + let Some(probe) = compile_offsets_for_times_probe() else { + return; + }; + let probe_dir = probe.path().to_str().expect("probe path is utf-8"); + let consumer = run_client( + KAFKA_IMAGE, + &[ + "java", + "-cp", + "/opt/kafka/libs/*:/probe", + "OffsetsForTimesProbe", + &gateway.to_string(), + TOPIC, + "0", + &between.to_string(), + "0", + &future.to_string(), + ], + &[(probe_dir, "/probe")], + ) + .expect_ran("java consumer offsetsForTimes"); + for expected in [ + format!("time={between} offset=3"), + "time=0 offset=0".to_string(), + format!("time={future} offset=null"), + ] { + assert!( + consumer.lines().any(|line| line.trim() == expected), + "offsetsForTimes must answer `{expected}`, got: {consumer}" + ); + } +} diff --git a/gateways/kafka/tests/list_offsets_real_bridge_tests.rs b/gateways/kafka/tests/list_offsets_real_bridge_tests.rs index 25c28c4fc..d1f6c27ef 100644 --- a/gateways/kafka/tests/list_offsets_real_bridge_tests.rs +++ b/gateways/kafka/tests/list_offsets_real_bridge_tests.rs @@ -22,17 +22,21 @@ //! `broker`-feature-only: request `Decodable` and response `Encodable`, not the reverse). use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bytes::Bytes; use iggy::prelude::{Identifier, IggyMessage, MessageClient, Partitioning}; use serial_test::serial; +use tokio::io::copy_bidirectional; +use tokio::net::{TcpListener, TcpStream}; use tokio_util::sync::CancellationToken; use iggy_gateway_kafka::bridge::IggyBridge; use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; use iggy_gateway_kafka::protocol::api::{ BrokerAdvertise, ERROR_NONE, ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION, - ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, GatewayState, + GatewayState, }; use iggy_gateway_kafka::protocol::handlers::list_offsets; @@ -48,6 +52,9 @@ const REQUEST_VERSION: i16 = 6; const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024; const LATEST_TIMESTAMP: i64 = -1; const EARLIEST_TIMESTAMP: i64 = -2; +const MAX_TIMESTAMP: i64 = -3; +/// The first version that may ask for [`MAX_TIMESTAMP`]. Its request and response share v6's shape. +const MAX_TIMESTAMP_VERSION: i16 = 7; /// Builds a v6 flexible `ListOffsets` request body for one topic/partition. fn build_request(topic: &str, partition_index: i32, timestamp: i64) -> Bytes { @@ -71,6 +78,12 @@ fn build_request(topic: &str, partition_index: i32, timestamp: i64) -> Bytes { /// Decodes a v6 flexible `ListOffsets` response's first partition result into `(error_code, /// offset)`. fn decode_first_result(body: Bytes) -> (i16, i64) { + let (error_code, _timestamp, offset) = decode_first_entry(body); + (error_code, offset) +} + +/// As [`decode_first_result`], keeping the timestamp: `(error_code, timestamp, offset)`. +fn decode_first_entry(body: Bytes) -> (i16, i64, i64) { let mut d = Decoder::new(body); let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); let _topics_plus_one = d.read_varint().expect("topics array count"); @@ -78,9 +91,22 @@ fn decode_first_result(body: Bytes) -> (i16, i64) { let _partitions_plus_one = d.read_varint().expect("partitions array count"); let _partition_index = d.read_i32().expect("partition_index"); let error_code = d.read_i16().expect("error_code"); - let _timestamp = d.read_i64().expect("timestamp"); + let timestamp = d.read_i64().expect("timestamp"); let offset = d.read_i64().expect("offset"); - (error_code, offset) + (error_code, timestamp, offset) +} + +/// One lookup at `version`, answered as `(error_code, timestamp, offset)`. +async fn send_at( + state: &GatewayState, + version: i16, + topic: &str, + partition_index: i32, + timestamp: i64, +) -> (i16, i64, i64) { + let body = build_request(topic, partition_index, timestamp); + let outcome = list_offsets::handle(state, version, body).await; + decode_first_entry(outcome.expect_response("ListOffsets request always answers")) } async fn send( @@ -264,18 +290,304 @@ async fn a_nonexistent_topic_returns_unknown_topic_or_partition() { assert_eq!(error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); } +/// Milliseconds since the epoch, the unit Kafka timestamps use. +fn now_millis() -> i64 { + let elapsed = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("the clock is past the epoch"); + i64::try_from(elapsed.as_millis()).expect("now fits an i64") +} + +/// Sends `values` to partition 0 of `orders` in one batch, so they share one append time. +async fn produce_batch(server: &TestServer, values: &[&str]) { + produce_batch_to(server, 0, values).await; +} + +/// As [`produce_batch`], to `partition`. +async fn produce_batch_to(server: &TestServer, partition: u32, values: &[&str]) { + let raw = iggy_server::raw_client(server).await; + let mut messages: Vec<IggyMessage> = values + .iter() + .map(|value| IggyMessage::from(*value)) + .collect(); + raw.send_messages( + &Identifier::named("kafka").expect("valid stream name"), + &Identifier::named("orders").expect("valid topic name"), + &Partitioning::partition_id(partition), + &mut messages, + ) + .await + .expect("send a batch"); +} + +/// Two batches with a measured gap between their append times: offsets 0-1, then 2-4. Returns a +/// time inside the gap. +async fn two_batches_with_a_gap(server: &TestServer) -> i64 { + produce_batch(server, &["a", "b"]).await; + // Iggy stamps append time in microseconds and the lookup takes milliseconds, so the gap has to + // be wide enough that a millisecond inside it is after the first batch and before the second. + tokio::time::sleep(Duration::from_millis(50)).await; + let between = now_millis(); + tokio::time::sleep(Duration::from_millis(50)).await; + produce_batch(server, &["c", "d", "e"]).await; + between +} + +#[tokio::test] +#[serial] +async fn a_timestamp_resolves_to_the_first_record_appended_at_or_after_it() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let (state, seed) = connected_state(&server).await; + seed.ensure_stream_and_topic("orders", 1) + .await + .expect("seed the topic"); + let before = now_millis() - 60_000; + let between = two_batches_with_a_gap(&server).await; + + let (error_code, timestamp, offset) = + send_at(&state, REQUEST_VERSION, "orders", 0, between).await; + assert_eq!(error_code, ERROR_NONE); + assert_eq!(offset, 2, "the second batch starts at offset 2"); + assert!( + timestamp >= between, + "the record found was appended after the time asked for, got {timestamp} for {between}" + ); + + let (error_code, _, offset) = send_at(&state, REQUEST_VERSION, "orders", 0, before).await; + assert_eq!(error_code, ERROR_NONE); + assert_eq!(offset, 0, "a time before every record finds the first"); + + // AKHQ asks for time 0 to find where each partition starts. + let (error_code, _, offset) = send_at(&state, REQUEST_VERSION, "orders", 0, 0).await; + assert_eq!(error_code, ERROR_NONE); + assert_eq!(offset, 0); +} + +#[tokio::test] +#[serial] +async fn a_timestamp_after_every_record_is_minus_one_without_an_error() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let (state, seed) = connected_state(&server).await; + seed.ensure_stream_and_topic("orders", 1) + .await + .expect("seed the topic"); + produce_batch(&server, &["a", "b"]).await; + + let later = now_millis() + 3_600_000; + let (error_code, timestamp, offset) = + send_at(&state, REQUEST_VERSION, "orders", 0, later).await; + assert_eq!( + (error_code, timestamp, offset), + (ERROR_NONE, -1, -1), + "Kafka answers a time past the last record with no error and -1" + ); +} + +#[tokio::test] +#[serial] +async fn a_timestamp_on_an_empty_partition_is_minus_one_without_an_error() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let (state, seed) = connected_state(&server).await; + seed.ensure_stream_and_topic("orders", 1) + .await + .expect("seed the topic"); + + for (version, requested) in [(REQUEST_VERSION, 0), (MAX_TIMESTAMP_VERSION, MAX_TIMESTAMP)] { + let (error_code, timestamp, offset) = + send_at(&state, version, "orders", 0, requested).await; + assert_eq!((error_code, timestamp, offset), (ERROR_NONE, -1, -1)); + } +} + #[tokio::test] #[serial] -async fn an_arbitrary_timestamp_is_unsupported() { +async fn max_timestamp_resolves_to_the_last_record_at_any_version() { let data_dir = tempfile::tempdir().expect("tempdir"); let server = TestServer::spawn(data_dir.path()).await; let (state, seed) = connected_state(&server).await; seed.ensure_stream_and_topic("orders", 1) .await .expect("seed the topic"); + let between = two_batches_with_a_gap(&server).await; + + for version in [REQUEST_VERSION, MAX_TIMESTAMP_VERSION] { + let (error_code, timestamp, offset) = + send_at(&state, version, "orders", 0, MAX_TIMESTAMP).await; + assert_eq!(error_code, ERROR_NONE, "v{version}"); + assert_eq!(offset, 4, "v{version}: the last of five records"); + assert!( + timestamp >= between, + "v{version}: got {timestamp}, the last batch came after {between}" + ); + } +} + +/// A TCP relay to `server` that leaves the next connections [`Self::stall_next`] counts +/// unanswered, so a client on one waits in its handshake until it gives up. +struct StallingRelay { + address: String, + stall: Arc<AtomicUsize>, +} + +impl StallingRelay { + async fn start(server: &TestServer) -> Self { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind the relay"); + let address = listener.local_addr().expect("relay address").to_string(); + let upstream = server.test_config().address; + let stall = Arc::new(AtomicUsize::new(0)); + let stall_count = Arc::clone(&stall); + tokio::spawn(async move { + while let Ok((mut inbound, _)) = listener.accept().await { + if stall_count + .try_update(Ordering::SeqCst, Ordering::SeqCst, |left| { + left.checked_sub(1) + }) + .is_ok() + { + tokio::spawn(async move { + std::future::pending::<()>().await; + drop(inbound); + }); + continue; + } + let upstream = upstream.clone(); + tokio::spawn(async move { + if let Ok(mut outbound) = TcpStream::connect(&upstream).await { + let _ = copy_bidirectional(&mut inbound, &mut outbound).await; + } + }); + } + }); + Self { address, stall } + } + + fn stall_next(&self, connections: usize) { + self.stall.store(connections, Ordering::SeqCst); + } +} + +/// The first two lookups' read clients connect through a relay that never answers. Each lookup's +/// own budget ends it, and the lookup after them still answers. With only the request deadline +/// bounding them, the second stall would run the request out and the last lookup would answer 7. +#[tokio::test] +#[serial] +async fn a_stalled_lookup_leaves_the_later_partitions_of_the_request_their_answers() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let relay = StallingRelay::start(&server).await; + let mut config = server.test_config(); + config.address.clone_from(&relay.address); + let bridge = IggyBridge::connect(config) + .await + .expect("bridge should connect through the relay"); + let state = GatewayState::new( + BrokerAdvertise::default(), + Some(Arc::new(bridge)), + TEST_MAX_FRAME_SIZE, + false, + 0, + GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()), + ); + let seed = IggyBridge::connect(server.test_config()) + .await + .expect("seed bridge should connect to a ready server"); + seed.ensure_stream_and_topic("orders", 3) + .await + .expect("seed a 3-partition topic"); + for partition in 0..3 { + produce_batch_to(&server, partition, &["a", "b"]).await; + } + // Connects the probe client now, so the stalled connection is the first read slot's. + let (error_code, _) = send(&state, "orders", 0, LATEST_TIMESTAMP).await; + assert_eq!(error_code, ERROR_NONE); + + relay.stall_next(2); + let topics = [TopicRequest { + name: "orders", + partitions: &[(0, 0), (1, 0), (2, 0)], + }]; + let answers: Vec<(i32, i16, i64)> = send_multi(&state, &topics) + .await + .iter() + .map(|(_, partition, error_code, offset)| (*partition, *error_code, *offset)) + .collect(); + assert_eq!( + answers, + vec![ + (0, ERROR_REQUEST_TIMED_OUT, -1), + (1, ERROR_REQUEST_TIMED_OUT, -1), + (2, ERROR_NONE, 0), + ] + ); +} + +/// Every lookup by time in one request shares one read slot, handed back after each lookup. If +/// it were not, every partition after the first would answer 7. +#[tokio::test] +#[serial] +async fn several_partitions_looked_up_by_time_in_one_request_each_resolve() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let (state, seed) = connected_state(&server).await; + seed.ensure_stream_and_topic("orders", 2) + .await + .expect("seed a 2-partition topic"); + produce_batch_to(&server, 0, &["a", "b"]).await; + produce_batch_to(&server, 1, &["x"]).await; + tokio::time::sleep(Duration::from_millis(50)).await; + let between = now_millis(); + tokio::time::sleep(Duration::from_millis(50)).await; + produce_batch_to(&server, 0, &["c"]).await; + produce_batch_to(&server, 1, &["y", "z"]).await; + + let topics = [TopicRequest { + name: "orders", + partitions: &[ + (0, between), + (1, between), + (1, 0), + (0, LATEST_TIMESTAMP), + (0, between), + ], + }]; + let results = send_multi(&state, &topics).await; + let answers: Vec<(i32, i16, i64)> = results + .iter() + .map(|(_, partition, error_code, offset)| (*partition, *error_code, *offset)) + .collect(); + assert_eq!( + answers, + vec![ + (0, ERROR_NONE, 2), + (1, ERROR_NONE, 1), + (1, ERROR_NONE, 0), + (0, ERROR_NONE, 3), + (0, ERROR_NONE, 2), + ], + "each lookup by time resolves on its own partition, in request order, and a repeated \ + entry gets the same answer" + ); +} - let (error_code, _) = send(&state, "orders", 0, 1_700_000_000_000).await; - assert_eq!(error_code, ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT); +#[tokio::test] +#[serial] +async fn a_timestamp_on_a_missing_partition_or_topic_is_unknown() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let (state, seed) = connected_state(&server).await; + seed.ensure_stream_and_topic("orders", 1) + .await + .expect("seed the topic"); + + let (error_code, _, _) = send_at(&state, REQUEST_VERSION, "orders", 5, 0).await; + assert_eq!(error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); + let (error_code, _, _) = send_at(&state, REQUEST_VERSION, "missing", 0, 0).await; + assert_eq!(error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION); } /// Regression test: a topic named in two separate request entries must still resolve every diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs index e0dce7872..911feae00 100644 --- a/gateways/kafka/tests/version_firewall_tests.rs +++ b/gateways/kafka/tests/version_firewall_tests.rs @@ -322,7 +322,7 @@ async fn create_topics_unsupported_version_above_max_closes_connection() { #[tokio::test] async fn list_offsets_unsupported_version_above_max_closes_connection() { assert!( - handle_request(API_KEY_LIST_OFFSETS, 7, Bytes::new(), &default_broker()) + handle_request(API_KEY_LIST_OFFSETS, 8, Bytes::new(), &default_broker()) .await .is_close(), "ListOffsets above encoder max must close rather than return a clamped body" @@ -792,12 +792,12 @@ async fn produce_advertises_min_zero_but_firewall_rejects_below_v3() { // ── Unsupported-version e2e paths for remaining scoped APIs ───────────────── #[tokio::test] -async fn list_offsets_v7_unsupported_e2e_closes_connection() { +async fn list_offsets_v8_unsupported_e2e_closes_connection() { let (addr, _shutdown) = spawn_test_server().await; let mut stream = TcpStream::connect(addr).await.expect("connect"); let frame = build_request_frame( API_KEY_LIST_OFFSETS, - 7, + 8, 370, Some("scope-test"), &[0x00, 0x00, 0x00, 0x00], @@ -806,7 +806,7 @@ async fn list_offsets_v7_unsupported_e2e_closes_connection() { assert_eq!( read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, ByteRead::Closed, - "ListOffsets v7 is above encoder max and must close" + "ListOffsets v8 is above the advertised max and must close" ); }
