feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge (#4259)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 271619d..c18175a 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -56,7 +56,7 @@
max-threads = 4
[[profile.default.overrides]]
-filter = 'binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)'
+filter = 'binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests) or binary_id(iggy-gateway-kafka::list_offsets_real_bridge_tests)'
test-group = "kafka_bridge"
[profile.default]
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 13ad306..66b5094 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -2,7 +2,13 @@
Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests, validates scoped API keys and versions, and returns stub responses.
-> **Stub warning:** no API persists or reads real data yet. Produce, Fetch, and ListOffsets return retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere instead of trusting a fake success. CreateTopics does **not** create topics; valid requests return `NOT_CONTROLLER` (41). Metadata still reports requested topics as unknown. Persistence lands with the Iggy bridge (see [docs/SCOPE.md](docs/SCOPE.md)).
+> **Stub warning:** most APIs still don't persist or read real data. Produce and Fetch return
+> retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere instead of
+> trusting a fake success. CreateTopics does **not** create topics; valid requests return
+> `NOT_CONTROLLER` (41). Metadata still reports requested topics as unknown. ListOffsets is wired
+> to the Iggy bridge: with `IGGY_KAFKA_BRIDGE_ENABLED=true` it answers `EARLIEST`/`LATEST` from
+> real partition state; with the bridge off (the default) it stays a stub and answers
+> `NOT_LEADER_OR_FOLLOWER` (6). See [docs/SCOPE.md](docs/SCOPE.md).
## Run
@@ -111,11 +117,11 @@
does on expiry - cannot abort it mid-flight. A timed-out call can leave that task holding the
shared client's connection lock for up to another 30s (the SDK's own reply deadline), queuing
every other bridge call behind it. `IggyBridge` holds one `IggyClient` with no pooling (see
-Concurrency ceiling below), so this only matters once concurrent Kafka connections share a bridge
-
-- tolerable today only because nothing calls this bridge from a live handler yet; must be resolved
-before `#3535`/`#3536`. See `IggyBridge`'s own doc comment (its rustdoc is private, so this isn't
-a followable link outside the crate - read the source at `src/bridge/iggy_bridge.rs`).
+Concurrency ceiling below) and no semaphore bounding concurrent bridge calls - no longer a
+hypothetical now that ListOffsets (`#3537`) calls it from a real handler; more pressing once
+CreateTopics (`#3538`), Metadata (`#3534`), Produce (`#3535`) and Fetch (`#3536`) add their own
+concurrent callers. See `IggyBridge`'s own doc comment (its rustdoc is private, so this isn't a
+followable link outside the crate - read the source at `src/bridge/iggy_bridge.rs`).
### Topic mapping
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 30289a9..ea05eda 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -113,6 +113,33 @@
- Iggy partitions are **0-based** (same as Kafka) — direct `partition_id` mapping, no offset conversion
- Kafka consumer groups do **not** map onto Iggy consumer groups. Assignment stays client-side, and Iggy's group registry is used as an offset key only ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md))
- `Partitioning::partition_id(index)` on every Produce. A Kafka producer resolves the partition before it builds the request, so `Partitioning::balanced()` has no trigger there. The `-1` default-partition-count case belongs to CreateTopics
+- [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`.
+ `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
+ tracks no rolling low-watermark distinct from partition creation, so once a partition is
+ old enough for retention to purge its first segment, `0` names a log-start offset that no
+ longer exists - a real consumer with `auto.offset.reset=earliest` seeks into a hole. Not
+ fixable client-side; needs the bridge to expose a real start offset. Harmless *today* only
+ because Fetch (`#3536`) is still a stub - nothing yet reads at the offset this returns.
+ - 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
+ 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`.
- [ ] Real Metadata topology (brokers, partitions, leaders) backed by Iggy state
### `kafka-protocol` crate adoption — superseded, done differently
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index e110791..5d388ff 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -72,11 +72,13 @@
/// so no finite value here can guarantee catching it. That same unboundedness is why this timeout
/// cannot simply be dropped for post-connect calls either: [`IggyBridge`] holds one `IggyClient`
/// with no pooling, so an unbounded reconnect dial with nothing here to stop it would wedge every
-/// later call on this bridge, not just the one that triggered it. Tolerable only because nothing
-/// calls this bridge from a live Kafka handler yet - closing it needs either a cooperatively
+/// later call on this bridge, not just the one that triggered it. No longer a hypothetical: since
+/// `ListOffsets` (#3537) this bridge is called from a live Kafka handler, with no semaphore
+/// bounding concurrent bridge calls. Closing the underlying gap needs either a cooperatively
/// cancellable SDK call or a deadline on the SDK's own reconnect dial, neither of which this
-/// bridge can add from the outside. Must be resolved before #3535/#3536 share this client across
-/// concurrent connections.
+/// bridge can add from the outside; bounding concurrent bridge calls is a separate, addressable
+/// fix that becomes more pressing as `CreateTopics` and `Metadata` (#3538/#3534) add more live
+/// callers.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
/// Wraps a single Iggy client call in [`REQUEST_TIMEOUT`]. See that constant's doc for why every
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index adf815e..3c29a95 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -75,6 +75,13 @@
/// `CreateTopics` stub: do not claim topics were created (no controller / no Iggy bridge).
pub const ERROR_NOT_CONTROLLER: i16 = 41;
pub const ERROR_INVALID_REQUEST: i16 = 42;
+/// `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 = 43;
/// Result of handling one Kafka request body.
#[derive(Debug)]
diff --git a/gateways/kafka/src/protocol/handlers/list_offsets.rs b/gateways/kafka/src/protocol/handlers/list_offsets.rs
index acc29a3..ef9b2d5 100644
--- a/gateways/kafka/src/protocol/handlers/list_offsets.rs
+++ b/gateways/kafka/src/protocol/handlers/list_offsets.rs
@@ -17,19 +17,29 @@
//! `ListOffsets` (API key 2).
+use std::collections::{HashMap, HashSet};
+use std::time::Duration;
+
use bytes::Bytes;
+use kafka_protocol::messages::list_offsets_request::{ListOffsetsPartition, ListOffsetsTopic};
use kafka_protocol::messages::list_offsets_response::{
ListOffsetsPartitionResponse, ListOffsetsTopicResponse,
};
use kafka_protocol::messages::{ListOffsetsRequest, ListOffsetsResponse};
+use tokio::time::Instant;
+use crate::bridge::{BridgeError, IggyBridge};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, GatewayState,
- HandleOutcome,
+ API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_NOT_LEADER_OR_FOLLOWER,
+ ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
};
use crate::protocol::bounds_guard::validate_list_offsets_shape;
-use crate::protocol::handlers::{decode_guarded, encode_message, handle_versioned_request};
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, handle_versioned_request, is_supported_version,
+ respond_or_close, unsupported_version_response,
+};
pub const RANGE: ApiVersionRange = ApiVersionRange {
api_key: API_KEY_LIST_OFFSETS,
@@ -37,24 +47,325 @@
max_version: 6,
};
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until a handler awaits the bridge"
-)]
+/// 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
+/// 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
+/// client that retries only its still-erroring topics (the common case) narrows below the cap on
+/// its own within a couple of retries, rather than resending the same oversized request forever.
+const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
+
+/// Wall-clock ceiling for one request's aggregate bridge work.
+///
+/// `ListOffsets` carries no `timeout_ms` field in any version this gateway supports (that field
+/// is v10+; [`RANGE`] tops out at v6) - unlike `CreateTopics`, there is no client-supplied value
+/// to honor here, so this is a fixed ceiling instead. Sized well above one `high_watermarks`
+/// call's own `REQUEST_TIMEOUT` (15s, bridge-internal) so a single slow-but-alive call is not the
+/// common trigger, while still bounding the sum across up to [`MAX_BRIDGE_BACKED_TOPICS`] calls -
+/// without this, a large batch against a struggling bridge could hold the shared client for
+/// `MAX_BRIDGE_BACKED_TOPICS * 15s`, not just one call's worth.
+///
+/// Applied per call, not once around the whole batch: [`resolve_all_topics`] checks it before
+/// starting each topic's `high_watermarks` call and wraps the call itself in
+/// [`tokio::time::timeout_at`] against the same instant, so a topic already resolved when the
+/// deadline arrives keeps its real answer and only the not-yet-started ones fall back to
+/// [`ERROR_REQUEST_TIMED_OUT`].
+const REQUEST_DEADLINE: Duration = Duration::from_secs(20);
+
+/// KIP-79 sentinel: the offset of the next message that would be produced.
+const LATEST_TIMESTAMP: i64 = -1;
+/// KIP-79 sentinel: the offset of the first message still retained.
+const EARLIEST_TIMESTAMP: i64 = -2;
+/// Placeholder offset/timestamp for a partition result that carries an error - matches real
+/// Kafka's own convention on the error path.
+const NO_OFFSET: i64 = -1;
+
+/// [`IggyBridge::high_watermarks`]'s return type, spelled once for [`resolve_one_partition`].
+type HighWatermarksResult =
+ core::result::Result<Vec<(u32, core::result::Result<i64, BridgeError>)>, BridgeError>;
+
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
- handle_versioned_request(
- API_KEY_LIST_OFFSETS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<ListOffsetsRequest>(v, b, |v, b| {
- validate_list_offsets_shape(v, b, state.max_frame_size)
- })
- },
- encode_response,
- encode_error_response,
- "ListOffsets",
- )
+ let Some(bridge) = &state.bridge else {
+ return handle_versioned_request(
+ API_KEY_LIST_OFFSETS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<ListOffsetsRequest>(v, b, |v, b| {
+ validate_list_offsets_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "ListOffsets",
+ );
+ };
+
+ if !is_supported_version(API_KEY_LIST_OFFSETS, api_version) {
+ return unsupported_version_response(API_KEY_LIST_OFFSETS, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+
+ let req = match decode_guarded::<ListOffsetsRequest>(api_version, body, |v, b| {
+ validate_list_offsets_shape(v, b, state.max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode ListOffsets request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "ListOffsets",
+ );
+ }
+ };
+
+ let deadline = Instant::now() + REQUEST_DEADLINE;
+ let topics = resolve_all_topics(bridge, &req.topics, deadline).await;
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// 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![])`
+ /// as far as [`resolve_one_partition`] is concerned (every partition in it fails its own
+ /// `u32::try_from` before ever consulting this).
+ Watermarks(HighWatermarksResult),
+ /// Beyond [`MAX_BRIDGE_BACKED_TOPICS`], or the deadline elapsed before this topic's turn - no
+ /// call was made. Answered [`ERROR_REQUEST_TIMED_OUT`] (retriable) rather than
+ /// [`ERROR_INVALID_REQUEST`] so a client's own per-topic retry narrows the batch on its own.
+ NotAttempted,
+}
+
+/// Dedupes `requested` by topic name, merging every entry's partitions.
+///
+/// `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.
+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();
+ for topic in requested {
+ let name = topic.name.as_str();
+ if !partitions_by_name.contains_key(name) {
+ order.push(name);
+ }
+ let entry = partitions_by_name.entry(name).or_default();
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ entry.extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+ (order, partitions_by_name)
+}
+
+/// Resolves one [`TopicLookup`] per name in `order`: [`TopicLookup::NotAttempted`] beyond
+/// [`MAX_BRIDGE_BACKED_TOPICS`] or once `deadline` has passed, otherwise a real `high_watermarks`
+/// 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
+/// partition's answer, since every one of them already fails its own index check.
+async fn resolve_topic_lookups<'a>(
+ bridge: &IggyBridge,
+ order: &[&'a str],
+ partitions_by_name: &HashMap<&'a str, Vec<u32>>,
+ deadline: Instant,
+) -> HashMap<&'a str, TopicLookup> {
+ let accepted: HashSet<&str> = order
+ .iter()
+ .take(MAX_BRIDGE_BACKED_TOPICS)
+ .copied()
+ .collect();
+ if order.len() > MAX_BRIDGE_BACKED_TOPICS {
+ // debug!, not warn!: the client controls how many topics it batches into one request and
+ // the connection stays open, so a consumer stuck above the cap logs this every retry.
+ tracing::debug!(
+ distinct_topics = order.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request exceeds the per-request topic cap; resolving the first {} and \
+ answering the rest retriable",
+ MAX_BRIDGE_BACKED_TOPICS
+ );
+ }
+
+ let mut lookups: HashMap<&str, TopicLookup> = HashMap::new();
+ let mut deadline_exceeded = false;
+ for &name in order {
+ if !accepted.contains(name) {
+ lookups.insert(name, TopicLookup::NotAttempted);
+ continue;
+ }
+ let partitions = partitions_by_name[name].as_slice();
+ if partitions.is_empty() {
+ lookups.insert(name, TopicLookup::Watermarks(Ok(Vec::new())));
+ continue;
+ }
+ if deadline_exceeded || Instant::now() >= deadline {
+ if !deadline_exceeded {
+ deadline_exceeded = true;
+ tracing::warn!(
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its deadline; \
+ answering remaining topics retriable instead of starting new bridge calls"
+ );
+ }
+ lookups.insert(name, TopicLookup::NotAttempted);
+ 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 \
+ deadline; answering retriable instead of blocking further"
+ );
+ lookups.insert(name, TopicLookup::NotAttempted);
+ continue;
+ }
+ };
+
+ if let Err(call_err) = &result {
+ let kafka_code = call_err.to_kafka_error_code();
+ if kafka_code == ERROR_UNKNOWN_TOPIC_OR_PARTITION {
+ // Client-caused (topic doesn't exist / isn't mapped): expected traffic, not
+ // operator-actionable.
+ tracing::debug!(topic = name, %call_err, "ListOffsets bridge lookup: topic not found");
+ } else {
+ tracing::error!(topic = name, %call_err, "ListOffsets bridge lookup failed");
+ }
+ }
+ lookups.insert(name, TopicLookup::Watermarks(result));
+ }
+ lookups
+}
+
+/// 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.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+ deadline: Instant,
+) -> Vec<ListOffsetsTopicResponse> {
+ let (order, partitions_by_name) = group_requested_topics(requested);
+ let lookups = resolve_topic_lookups(bridge, &order, &partitions_by_name, deadline).await;
+
+ requested
+ .iter()
+ .map(|topic| {
+ // Always present: `order` (and so `lookups`) was built from exactly these same
+ // requested topic names, just above.
+ let lookup = lookups
+ .get(topic.name.as_str())
+ .expect("every requested topic name was resolved above");
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|requested| resolve_one_partition(requested, lookup))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+ })
+ .collect()
+}
+
+/// `lookup` is the whole topic's resolution outcome: an errored [`TopicLookup::Watermarks`] 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);
+ };
+
+ let results = match lookup {
+ TopicLookup::NotAttempted => {
+ return error_response(requested.partition_index, ERROR_REQUEST_TIMED_OUT);
+ }
+ TopicLookup::Watermarks(Err(call_err)) => {
+ return error_response(requested.partition_index, call_err.to_kafka_error_code());
+ }
+ TopicLookup::Watermarks(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),
+ // 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 (_, watermark) = &results[found];
+
+ let watermark = match watermark {
+ Err(err) => return error_response(requested.partition_index, err.to_kafka_error_code()),
+ Ok(watermark) => *watermark,
+ };
+
+ match requested.timestamp {
+ LATEST_TIMESTAMP => offset_response(requested.partition_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. Harmless *today*
+ // only because Fetch (`#3536`) is still a stub - nothing yet reads at the offset this
+ // returns. Not fixable client-side; needs the bridge to expose a real start offset.
+ 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,
+ ),
+ }
+}
+
+fn offset_response(partition: i32, offset: i64) -> ListOffsetsPartitionResponse {
+ ListOffsetsPartitionResponse::default()
+ .with_partition_index(partition)
+ .with_timestamp(LATEST_TIMESTAMP)
+ .with_offset(offset)
+}
+
+fn error_response(partition: i32, error_code: i16) -> ListOffsetsPartitionResponse {
+ ListOffsetsPartitionResponse::default()
+ .with_partition_index(partition)
+ .with_error_code(error_code)
+ .with_timestamp(NO_OFFSET)
+ .with_offset(NO_OFFSET)
}
/// Well-formed `ListOffsets` response with a single placeholder topic/partition.
@@ -71,7 +382,7 @@
pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
let topics = vec![
ListOffsetsTopicResponse::default()
- .with_partitions(vec![partition_response(0, error_code)]),
+ .with_partitions(vec![stub_partition_response(0, error_code)]),
];
encode_inner(version, topics)
}
@@ -94,7 +405,7 @@
.partitions
.iter()
.map(|p| {
- partition_response(p.partition_index, ERROR_NOT_LEADER_OR_FOLLOWER)
+ stub_partition_response(p.partition_index, ERROR_NOT_LEADER_OR_FOLLOWER)
})
.collect(),
)
@@ -108,8 +419,105 @@
encode_message(&resp, version, 256)
}
-fn partition_response(partition: i32, error_code: i16) -> ListOffsetsPartitionResponse {
+fn stub_partition_response(partition: i32, error_code: i16) -> ListOffsetsPartitionResponse {
ListOffsetsPartitionResponse::default()
.with_partition_index(partition)
.with_error_code(error_code)
}
+
+#[cfg(test)]
+mod tests {
+ use crate::protocol::api::ERROR_NONE;
+
+ use super::*;
+
+ 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()
+ }
+
+ #[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);
+ 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);
+ 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, NO_OFFSET);
+ }
+
+ #[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());
+ }
+
+ #[test]
+ fn a_topic_beyond_the_cap_or_past_the_deadline_answers_retriable() {
+ let resp =
+ resolve_one_partition(&partition(0, LATEST_TIMESTAMP), &TopicLookup::NotAttempted);
+ assert_eq!(resp.error_code, ERROR_REQUEST_TIMED_OUT);
+ assert_eq!(resp.offset, NO_OFFSET);
+ }
+
+ #[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);
+ 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![(
+ 0,
+ Err(BridgeError::PartitionOutOfRange {
+ topic: "orders".to_string(),
+ partition: 0,
+ partitions_count: 0,
+ }),
+ )]));
+ let resp = resolve_one_partition(&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());
+ }
+
+ #[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);
+ assert_eq!(resp.error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ }
+}
diff --git a/gateways/kafka/tests/list_offsets_real_bridge_tests.rs b/gateways/kafka/tests/list_offsets_real_bridge_tests.rs
new file mode 100644
index 0000000..fbbcdae
--- /dev/null
+++ b/gateways/kafka/tests/list_offsets_real_bridge_tests.rs
@@ -0,0 +1,401 @@
+// 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.
+
+//! Wire-level `ListOffsets` tests against a real `iggy-server` process, through
+//! [`list_offsets::handle`] with a connected [`GatewayState`]. See
+//! `create_topics_real_bridge_tests.rs` for why requests/responses are hand-built here rather
+//! than through `kafka_protocol`'s own `Encodable`/`Decodable` (this crate builds
+//! `broker`-feature-only: request `Decodable` and response `Encodable`, not the reverse).
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use iggy::prelude::{Identifier, IggyMessage, MessageClient, Partitioning};
+use serial_test::serial;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::protocol::api::{
+ BrokerAdvertise, ERROR_NONE, ERROR_REQUEST_TIMED_OUT, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, GatewayState,
+};
+use iggy_gateway_kafka::protocol::handlers::list_offsets;
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+
+use codec::{Decoder, Encoder};
+use iggy_server::TestServer;
+
+const REQUEST_VERSION: i16 = 6;
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+const LATEST_TIMESTAMP: i64 = -1;
+const EARLIEST_TIMESTAMP: i64 = -2;
+
+/// Builds a v6 flexible `ListOffsets` request body for one topic/partition.
+fn build_request(topic: &str, partition_index: i32, timestamp: i64) -> Bytes {
+ let mut enc = Encoder::with_capacity(128);
+ enc.write_i32(-1); // replica_id: ordinary client, not a follower broker
+ enc.write_i8(0); // isolation_level: READ_UNCOMMITTED
+
+ enc.write_varint(2); // one topic
+ enc.write_compact_nullable_string(Some(topic));
+ enc.write_varint(2); // one partition
+ enc.write_i32(partition_index);
+ enc.write_i32(-1); // current_leader_epoch: unset
+ enc.write_i64(timestamp);
+ enc.write_empty_tagged_fields(); // partition tagged fields
+ enc.write_empty_tagged_fields(); // topic tagged fields
+
+ enc.write_empty_tagged_fields(); // top-level tagged fields
+ enc.freeze()
+}
+
+/// Decodes a v6 flexible `ListOffsets` response's first partition result into `(error_code,
+/// offset)`.
+fn decode_first_result(body: Bytes) -> (i16, 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");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ 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 offset = d.read_i64().expect("offset");
+ (error_code, offset)
+}
+
+async fn send(
+ state: &GatewayState,
+ topic: &str,
+ partition_index: i32,
+ timestamp: i64,
+) -> (i16, i64) {
+ let body = build_request(topic, partition_index, timestamp);
+ let outcome = list_offsets::handle(state, REQUEST_VERSION, body).await;
+ let resp_body = outcome.expect_response("ListOffsets request always answers");
+ decode_first_result(resp_body)
+}
+
+/// One requested topic entry, for [`build_multi_request`]: a name and its `(partition_index,
+/// timestamp)` pairs.
+struct TopicRequest<'a> {
+ name: &'a str,
+ partitions: &'a [(i32, i64)],
+}
+
+/// Builds a v6 flexible `ListOffsets` request body for several topic entries at once - unlike
+/// [`build_request`], `topics` may repeat the same name across more than one entry.
+fn build_multi_request(topics: &[TopicRequest]) -> Bytes {
+ let mut enc = Encoder::with_capacity(4096);
+ enc.write_i32(-1); // replica_id
+ enc.write_i8(0); // isolation_level
+
+ enc.write_varint((topics.len() + 1) as u64);
+ for topic in topics {
+ enc.write_compact_nullable_string(Some(topic.name));
+ enc.write_varint((topic.partitions.len() + 1) as u64);
+ for &(partition_index, timestamp) in topic.partitions {
+ enc.write_i32(partition_index);
+ enc.write_i32(-1); // current_leader_epoch
+ enc.write_i64(timestamp);
+ enc.write_empty_tagged_fields();
+ }
+ enc.write_empty_tagged_fields();
+ }
+ enc.write_empty_tagged_fields();
+ enc.freeze()
+}
+
+/// Decodes every topic/partition result in a v6 flexible `ListOffsets` response into `(name,
+/// partition_index, error_code, offset)`, in wire order.
+fn decode_all(body: Bytes) -> Vec<(String, i32, i16, 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");
+ let mut results = Vec::new();
+ for _ in 1..topics_plus_one {
+ let name = d
+ .read_compact_nullable_string()
+ .expect("topic name")
+ .expect("name is never null in a request-echoing response");
+ let partitions_plus_one = d.read_varint().expect("partitions array count");
+ for _ in 1..partitions_plus_one {
+ 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 offset = d.read_i64().expect("offset");
+ let _leader_epoch = d.read_i32().expect("leader_epoch");
+ let _partition_tagged_fields = d.read_varint().expect("partition tagged fields");
+ results.push((name.clone(), partition_index, error_code, offset));
+ }
+ let _topic_tagged_fields = d.read_varint().expect("topic tagged fields");
+ }
+ results
+}
+
+async fn send_multi(
+ state: &GatewayState,
+ topics: &[TopicRequest<'_>],
+) -> Vec<(String, i32, i16, i64)> {
+ let body = build_multi_request(topics);
+ let outcome = list_offsets::handle(state, REQUEST_VERSION, body).await;
+ let resp_body = outcome.expect_response("ListOffsets request always answers");
+ decode_all(resp_body)
+}
+
+async fn connected_state(server: &TestServer) -> (GatewayState, IggyBridge) {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ // A second bridge for direct seeding (get_topics/create/produce) alongside the handler's own.
+ let seed_bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge should connect to a ready server");
+ let state = GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ );
+ (state, seed_bridge)
+}
+
+#[tokio::test]
+#[serial]
+async fn latest_on_a_fresh_empty_partition_is_zero() {
+ 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, offset) = send(&state, "orders", 0, LATEST_TIMESTAMP).await;
+ assert_eq!(error_code, ERROR_NONE);
+ assert_eq!(offset, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn latest_reflects_produced_messages() {
+ 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 raw = iggy_server::raw_client(&server).await;
+ let mut messages: Vec<IggyMessage> = (0..3)
+ .map(|i| IggyMessage::from(format!("message-{i}")))
+ .collect();
+ raw.send_messages(
+ &Identifier::named("kafka").expect("valid stream name"),
+ &Identifier::named("orders").expect("valid topic name"),
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("seed 3 messages");
+
+ let (error_code, offset) = send(&state, "orders", 0, LATEST_TIMESTAMP).await;
+ assert_eq!(error_code, ERROR_NONE);
+ assert_eq!(offset, 3);
+}
+
+#[tokio::test]
+#[serial]
+async fn earliest_is_always_zero() {
+ 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, offset) = send(&state, "orders", 0, EARLIEST_TIMESTAMP).await;
+ assert_eq!(error_code, ERROR_NONE);
+ assert_eq!(offset, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn out_of_range_partition_returns_unknown_topic_or_partition() {
+ 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 a 1-partition topic");
+
+ let (error_code, _) = send(&state, "orders", 5, LATEST_TIMESTAMP).await;
+ assert_eq!(error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+}
+
+#[tokio::test]
+#[serial]
+async fn a_nonexistent_topic_returns_unknown_topic_or_partition() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (state, _seed) = connected_state(&server).await;
+
+ let (error_code, _) = send(&state, "orders", 0, LATEST_TIMESTAMP).await;
+ assert_eq!(error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+}
+
+#[tokio::test]
+#[serial]
+async fn an_arbitrary_timestamp_is_unsupported() {
+ 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(&state, "orders", 0, 1_700_000_000_000).await;
+ assert_eq!(error_code, ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT);
+}
+
+/// Regression test: a topic named in two separate request entries must still resolve every
+/// partition across both correctly, not just avoid a crash - proves the dedup-by-name merge
+/// actually unions the two entries' partition lists rather than dropping one.
+#[tokio::test]
+#[serial]
+async fn a_topic_named_in_two_request_entries_resolves_both_entries_correctly() {
+ 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");
+
+ let topics = [
+ TopicRequest {
+ name: "orders",
+ partitions: &[(0, LATEST_TIMESTAMP)],
+ },
+ TopicRequest {
+ name: "orders",
+ partitions: &[(1, LATEST_TIMESTAMP)],
+ },
+ ];
+ let results = send_multi(&state, &topics).await;
+ assert_eq!(results.len(), 2);
+ for (name, _partition_index, error_code, offset) in &results {
+ assert_eq!(name, "orders");
+ assert_eq!(*error_code, ERROR_NONE);
+ assert_eq!(*offset, 0);
+ }
+}
+
+/// Regression test: the binary search in `resolve_one_partition` must map each partition index to
+/// *its own* watermark, not just pass when both happen to agree. The two-empty-partitions test
+/// above can't catch a broken search or a constant-index bug - both partitions are 0 either way.
+#[tokio::test]
+#[serial]
+async fn distinct_partitions_resolve_to_their_own_distinct_watermarks() {
+ 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");
+
+ let raw = iggy_server::raw_client(&server).await;
+ let stream_id = Identifier::named("kafka").expect("valid stream name");
+ let topic_id = Identifier::named("orders").expect("valid topic name");
+ let mut partition_0: Vec<IggyMessage> = (0..3)
+ .map(|i| IggyMessage::from(format!("p0-{i}")))
+ .collect();
+ raw.send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut partition_0,
+ )
+ .await
+ .expect("seed partition 0 with 3 messages");
+ let mut partition_1: Vec<IggyMessage> = (0..7)
+ .map(|i| IggyMessage::from(format!("p1-{i}")))
+ .collect();
+ raw.send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(1),
+ &mut partition_1,
+ )
+ .await
+ .expect("seed partition 1 with 7 messages");
+
+ let topics = [TopicRequest {
+ name: "orders",
+ partitions: &[(1, LATEST_TIMESTAMP), (0, LATEST_TIMESTAMP)], // deliberately out of index order
+ }];
+ let results = send_multi(&state, &topics).await;
+ assert_eq!(results.len(), 2);
+ for (_, partition_index, error_code, offset) in &results {
+ assert_eq!(*error_code, ERROR_NONE);
+ let expected = if *partition_index == 0 { 3 } else { 7 };
+ assert_eq!(
+ *offset, expected,
+ "partition {partition_index} got the wrong watermark"
+ );
+ }
+}
+
+/// Regression test: a request naming more than the bridge-backed topic cap must be rejected
+/// wholesale (every entry, `INVALID_REQUEST`) rather than partially served or left unbounded.
+#[tokio::test]
+#[serial]
+async fn more_than_the_topic_cap_is_rejected() {
+ 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 one real, resolvable topic");
+
+ // 100 nonexistent names fill the cap; the one real, resolvable topic goes last so a correct
+ // cap leaves it untouched by this round entirely - proving the cap bounds *which* topics get
+ // a bridge call, not just that a request over the cap fails uniformly.
+ let mut names: Vec<String> = (0..100).map(|i| format!("topic-{i}")).collect();
+ names.push("orders".to_string());
+ let partitions = [(0, LATEST_TIMESTAMP)];
+ let topics: Vec<TopicRequest> = names
+ .iter()
+ .map(|name| TopicRequest {
+ name,
+ partitions: &partitions,
+ })
+ .collect();
+
+ let results = send_multi(&state, &topics).await;
+ assert_eq!(results.len(), 101);
+ for (name, _, error_code, _) in &results {
+ if name == "orders" {
+ assert_eq!(
+ *error_code, ERROR_REQUEST_TIMED_OUT,
+ "the 101st topic must be left unattempted by the cap, not looked up"
+ );
+ } else {
+ assert_eq!(*error_code, ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ }
+ }
+}
