numinnex commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4073059435
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,365 @@
//! `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,
min_version: 1,
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;
Review Comment:
This makes ListOffsets the first live handler to await the bridge; five
others still carry `reason = "kept until a handler awaits the bridge"`. Both
`README.md:122` and `iggy_bridge/mod.rs`'s `REQUEST_TIMEOUT` doc still say the
single unpooled `IggyClient` is "tolerable only because nothing calls this
bridge from a live Kafka handler yet ... must be resolved before #3535/#3536".
With 1024 connections at up to 100 calls each and no semaphore on bridge work,
this either wants a bound on concurrent bridge calls or those two claims
updated.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,365 @@
//! `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,
min_version: 1,
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.
+ Watermarks(HighWatermarksResult),
+ /// No partition of this topic asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`], so no
+ /// call was made at all - every partition here answers
+ /// [`ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT`] regardless of what a lookup
would have returned.
+ NoLookupNeeded,
+ /// 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 and
noting - per name -
+/// whether any partition asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`].
+///
+/// `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.
+fn group_requested_topics(
+ requested: &[ListOffsetsTopic],
+) -> (Vec<&str>, HashMap<&str, Vec<u32>>, HashSet<&str>) {
+ let mut order: Vec<&str> = Vec::new();
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ let mut needs_lookup: HashSet<&str> = HashSet::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();
+ for p in &topic.partitions {
+ if let Ok(index) = u32::try_from(p.partition_index) {
+ entry.push(index);
+ }
+ if matches!(p.timestamp, LATEST_TIMESTAMP | EARLIEST_TIMESTAMP) {
+ needs_lookup.insert(name);
+ }
+ }
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+ (order, partitions_by_name, needs_lookup)
+}
+
+/// Resolves one [`TopicLookup`] per name in `order`:
[`TopicLookup::NotAttempted`] beyond
+/// [`MAX_BRIDGE_BACKED_TOPICS`] or once `deadline` has passed,
[`TopicLookup::NoLookupNeeded`]
+/// when `needs_lookup` excludes the name, 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.
+async fn resolve_topic_lookups<'a>(
+ bridge: &IggyBridge,
+ order: &[&'a str],
+ partitions_by_name: &HashMap<&'a str, Vec<u32>>,
+ needs_lookup: &HashSet<&'a str>,
+ 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;
+ }
+ if !needs_lookup.contains(name) {
+ lookups.insert(name, TopicLookup::NoLookupNeeded);
+ 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 partitions = partitions_by_name[name].as_slice();
+ 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, needs_lookup) =
group_requested_topics(requested);
+ let lookups =
+ resolve_topic_lookups(bridge, &order, &partitions_by_name,
&needs_lookup, 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);
+ }
+ // By construction (`needs_lookup` in `resolve_all_topics`) every
partition of a topic in
+ // this state has a non-sentinel timestamp, so there is nothing to
branch on here.
+ TopicLookup::NoLookupNeeded => {
Review Comment:
Existence is only ever discovered by the `high_watermarks` call, which this
branch skips, so a topic that does not exist answers 43 instead of 3. Java's
`OffsetFetcherUtils` drops 43 with a `log.debug`, leaving the partition in
neither `fetchedOffsets` nor `partitionsToRetry`: no metadata refresh, no
retry, silent null. It also makes the code depend on request shape, since
adding a `(0, -1)` partition to the same entry returns 3 for the identical
occurrence. Resolving existence before branching on the timestamp would settle
both.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,365 @@
//! `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,
min_version: 1,
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.
+ Watermarks(HighWatermarksResult),
+ /// No partition of this topic asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`], so no
+ /// call was made at all - every partition here answers
+ /// [`ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT`] regardless of what a lookup
would have returned.
+ NoLookupNeeded,
+ /// 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 and
noting - per name -
+/// whether any partition asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`].
+///
+/// `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.
+fn group_requested_topics(
+ requested: &[ListOffsetsTopic],
+) -> (Vec<&str>, HashMap<&str, Vec<u32>>, HashSet<&str>) {
+ let mut order: Vec<&str> = Vec::new();
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ let mut needs_lookup: HashSet<&str> = HashSet::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();
+ for p in &topic.partitions {
+ if let Ok(index) = u32::try_from(p.partition_index) {
+ entry.push(index);
+ }
+ if matches!(p.timestamp, LATEST_TIMESTAMP | EARLIEST_TIMESTAMP) {
+ needs_lookup.insert(name);
Review Comment:
This insert keys off the timestamp alone, outside the `u32::try_from` guard
three lines up, so a topic whose sentinel partitions all carry negative indices
still reaches `high_watermarks(name, &[])`. That is a full `get_topic` round
trip nothing can consume, since `resolve_one_partition` rejects the negative
index before reading the result. `bounds_guard` doesn't check the sign, so one
small frame can buy up to 100 of these.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,365 @@
//! `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,
min_version: 1,
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.
+ Watermarks(HighWatermarksResult),
+ /// No partition of this topic asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`], so no
+ /// call was made at all - every partition here answers
+ /// [`ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT`] regardless of what a lookup
would have returned.
+ NoLookupNeeded,
+ /// 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 and
noting - per name -
+/// whether any partition asked for [`LATEST_TIMESTAMP`] or
[`EARLIEST_TIMESTAMP`].
+///
+/// `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.
+fn group_requested_topics(
+ requested: &[ListOffsetsTopic],
+) -> (Vec<&str>, HashMap<&str, Vec<u32>>, HashSet<&str>) {
+ let mut order: Vec<&str> = Vec::new();
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ let mut needs_lookup: HashSet<&str> = HashSet::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();
+ for p in &topic.partitions {
+ if let Ok(index) = u32::try_from(p.partition_index) {
+ entry.push(index);
+ }
+ if matches!(p.timestamp, LATEST_TIMESTAMP | EARLIEST_TIMESTAMP) {
+ needs_lookup.insert(name);
+ }
+ }
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+ (order, partitions_by_name, needs_lookup)
+}
+
+/// Resolves one [`TopicLookup`] per name in `order`:
[`TopicLookup::NotAttempted`] beyond
+/// [`MAX_BRIDGE_BACKED_TOPICS`] or once `deadline` has passed,
[`TopicLookup::NoLookupNeeded`]
+/// when `needs_lookup` excludes the name, 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.
+async fn resolve_topic_lookups<'a>(
+ bridge: &IggyBridge,
+ order: &[&'a str],
+ partitions_by_name: &HashMap<&'a str, Vec<u32>>,
+ needs_lookup: &HashSet<&'a str>,
+ 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;
+ }
+ if !needs_lookup.contains(name) {
+ lookups.insert(name, TopicLookup::NoLookupNeeded);
+ 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 partitions = partitions_by_name[name].as_slice();
+ 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, needs_lookup) =
group_requested_topics(requested);
+ let lookups =
+ resolve_topic_lookups(bridge, &order, &partitions_by_name,
&needs_lookup, 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);
+ }
+ // By construction (`needs_lookup` in `resolve_all_topics`) every
partition of a topic in
+ // this state has a non-sentinel timestamp, so there is nothing to
branch on here.
+ TopicLookup::NoLookupNeeded => {
+ return error_response(
+ requested.partition_index,
+ ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT,
+ );
+ }
+ 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 {
Review Comment:
Nothing tests this mapping. `latest_reflects_produced_messages` seeds a
1-partition topic, and the only multi-partition test uses two *empty*
partitions and asserts offset 0 for both, so no test anywhere has two
partitions at different watermarks. Replacing this search with a constant index
keeps all 293 tests green. Seeding partition 0 with 3 messages and partition 1
with 7, then asserting both in one request, would pin it.
##########
gateways/kafka/tests/list_offsets_real_bridge_tests.rs:
##########
@@ -0,0 +1,346 @@
+// 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;
Review Comment:
Each of the 8 tests in this file spawns a real `iggy-server`, but
`.config/nextest.toml:59` scopes the `kafka_bridge` group (`max-threads = 4`)
to `binary_id(iggy-gateway-kafka::bridge_iggy_integration_tests)` only, so this
binary runs outside that cap. `#[serial]` doesn't substitute under nextest's
process-per-test model. Adding this binary to that filter measured peak
concurrent servers 14 -> 6 at no wall-clock cost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
