krishvishal commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4071076193
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
Review Comment:
If the deadline fires, every topic that already resolved gets thrown away
too.
Anything that needs more than 20s then fails the same way on every retry.
Keep the resolved ones, wrap each call in `timeout_at(deadline, ..)` and
don't start new calls after it.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
+///
+/// `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - Iggy
exposes no
+/// per-message timestamp index, so "offset of the first message at or after
timestamp T" for any
+/// other `T` cannot be answered without one. Those partitions get
[`ERROR_UNKNOWN_SERVER_ERROR`],
+/// not a fabricated offset.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+) -> Vec<ListOffsetsTopicResponse> {
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ for topic in requested {
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ partitions_by_name
+ .entry(topic.name.as_str())
+ .or_default()
+ .extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+
+ let mut watermarks_by_name: HashMap<&str, HighWatermarksResult> =
HashMap::new();
+ for (&name, partitions) in &partitions_by_name {
+ let result = if partitions.is_empty() {
+ Ok(Vec::new())
+ } else {
+ bridge.high_watermarks(name, partitions).await
+ };
+ watermarks_by_name.insert(name, result);
+ }
+
+ requested
+ .iter()
+ .map(|topic| {
+ // Always present: partitions_by_name (and so watermarks_by_name)
was built from
+ // exactly these same requested topic names, just above.
+ let watermarks = watermarks_by_name
+ .get(topic.name.as_str())
+ .expect("every requested topic name was resolved above");
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|requested| resolve_one_partition(requested, watermarks))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+ })
+ .collect()
+}
+
+/// One response entry for `topic`, every requested partition carrying
`error_code` - used for a
+/// whole-request failure (over the topic cap, or the aggregate deadline)
where no bridge call
+/// was made for this topic at all.
+fn error_topic_response(topic: &ListOffsetsTopic, error_code: i16) ->
ListOffsetsTopicResponse {
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|p| error_response(p.partition_index, error_code))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+}
+
+/// `watermarks` is the whole topic's batched result: `Err` is a call-level
failure (e.g. the
+/// mapped stream doesn't exist) applying to every partition alike; the inner
per-partition
+/// `Result` inside `Ok` is [`BridgeError::PartitionOutOfRange`] for one bad
index among otherwise
+/// resolvable ones.
+fn resolve_one_partition(
+ requested: &ListOffsetsPartition,
+ watermarks: &HighWatermarksResult,
+) -> 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 watermarks {
+ Err(call_err) => {
+ return error_response(requested.partition_index,
call_err.to_kafka_error_code());
+ }
+ Ok(results) => results,
+ };
+
+ // Always present: `valid_partitions` (built from the same
`topic.partitions`) is exactly
+ // what was passed to `high_watermarks`, which returns one result per
requested partition.
+ let Some((_, watermark)) = results.iter().find(|(index, _)| *index ==
partition_index) else {
+ return error_response(requested.partition_index,
ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ };
+
+ 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),
+ _ => error_response(requested.partition_index,
ERROR_UNKNOWN_SERVER_ERROR),
Review Comment:
Java consumer retries -1 until default.api.timeout.ms, so offsetsForTimes
just spins and then times out.
UNSUPPORTED_FOR_MESSAGE_FORMAT (43) gives a null result without the retry
loop.
Worth a line in SCOPE.md too that by_duration resets won't work.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
+///
+/// `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - Iggy
exposes no
+/// per-message timestamp index, so "offset of the first message at or after
timestamp T" for any
+/// other `T` cannot be answered without one. Those partitions get
[`ERROR_UNKNOWN_SERVER_ERROR`],
+/// not a fabricated offset.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+) -> Vec<ListOffsetsTopicResponse> {
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ for topic in requested {
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ partitions_by_name
+ .entry(topic.name.as_str())
+ .or_default()
+ .extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+
+ let mut watermarks_by_name: HashMap<&str, HighWatermarksResult> =
HashMap::new();
+ for (&name, partitions) in &partitions_by_name {
+ let result = if partitions.is_empty() {
+ Ok(Vec::new())
+ } else {
+ bridge.high_watermarks(name, partitions).await
+ };
+ watermarks_by_name.insert(name, result);
+ }
+
+ requested
+ .iter()
+ .map(|topic| {
+ // Always present: partitions_by_name (and so watermarks_by_name)
was built from
+ // exactly these same requested topic names, just above.
+ let watermarks = watermarks_by_name
+ .get(topic.name.as_str())
+ .expect("every requested topic name was resolved above");
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|requested| resolve_one_partition(requested, watermarks))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+ })
+ .collect()
+}
+
+/// One response entry for `topic`, every requested partition carrying
`error_code` - used for a
+/// whole-request failure (over the topic cap, or the aggregate deadline)
where no bridge call
+/// was made for this topic at all.
+fn error_topic_response(topic: &ListOffsetsTopic, error_code: i16) ->
ListOffsetsTopicResponse {
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|p| error_response(p.partition_index, error_code))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+}
+
+/// `watermarks` is the whole topic's batched result: `Err` is a call-level
failure (e.g. the
+/// mapped stream doesn't exist) applying to every partition alike; the inner
per-partition
+/// `Result` inside `Ok` is [`BridgeError::PartitionOutOfRange`] for one bad
index among otherwise
+/// resolvable ones.
+fn resolve_one_partition(
+ requested: &ListOffsetsPartition,
+ watermarks: &HighWatermarksResult,
+) -> 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 watermarks {
+ Err(call_err) => {
+ return error_response(requested.partition_index,
call_err.to_kafka_error_code());
+ }
+ Ok(results) => results,
+ };
+
+ // Always present: `valid_partitions` (built from the same
`topic.partitions`) is exactly
+ // what was passed to `high_watermarks`, which returns one result per
requested partition.
+ let Some((_, watermark)) = results.iter().find(|(index, _)| *index ==
partition_index) else {
Review Comment:
Linear `find` per partition, so a request with 4095 partitions of one topic
is quadratic.
Roughly 10ms+ of blocked worker in a quick bench.
Partitions are sorted + deduped and high_watermarks keeps the order, so
`binary_search_by_key` works here.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
+///
+/// `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - Iggy
exposes no
+/// per-message timestamp index, so "offset of the first message at or after
timestamp T" for any
+/// other `T` cannot be answered without one. Those partitions get
[`ERROR_UNKNOWN_SERVER_ERROR`],
+/// not a fabricated offset.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+) -> Vec<ListOffsetsTopicResponse> {
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ for topic in requested {
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ partitions_by_name
+ .entry(topic.name.as_str())
+ .or_default()
+ .extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+
+ let mut watermarks_by_name: HashMap<&str, HighWatermarksResult> =
HashMap::new();
+ for (&name, partitions) in &partitions_by_name {
+ let result = if partitions.is_empty() {
+ Ok(Vec::new())
+ } else {
+ bridge.high_watermarks(name, partitions).await
+ };
+ watermarks_by_name.insert(name, result);
+ }
+
+ requested
+ .iter()
+ .map(|topic| {
+ // Always present: partitions_by_name (and so watermarks_by_name)
was built from
+ // exactly these same requested topic names, just above.
+ let watermarks = watermarks_by_name
+ .get(topic.name.as_str())
+ .expect("every requested topic name was resolved above");
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|requested| resolve_one_partition(requested, watermarks))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+ })
+ .collect()
+}
+
+/// One response entry for `topic`, every requested partition carrying
`error_code` - used for a
+/// whole-request failure (over the topic cap, or the aggregate deadline)
where no bridge call
+/// was made for this topic at all.
+fn error_topic_response(topic: &ListOffsetsTopic, error_code: i16) ->
ListOffsetsTopicResponse {
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|p| error_response(p.partition_index, error_code))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+}
+
+/// `watermarks` is the whole topic's batched result: `Err` is a call-level
failure (e.g. the
+/// mapped stream doesn't exist) applying to every partition alike; the inner
per-partition
+/// `Result` inside `Ok` is [`BridgeError::PartitionOutOfRange`] for one bad
index among otherwise
+/// resolvable ones.
+fn resolve_one_partition(
+ requested: &ListOffsetsPartition,
+ watermarks: &HighWatermarksResult,
+) -> 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 watermarks {
+ Err(call_err) => {
+ return error_response(requested.partition_index,
call_err.to_kafka_error_code());
+ }
+ Ok(results) => results,
+ };
+
+ // Always present: `valid_partitions` (built from the same
`topic.partitions`) is exactly
+ // what was passed to `high_watermarks`, which returns one result per
requested partition.
Review Comment:
nit: there's no `valid_partitions`, the list comes from partitions_by_name
which merges all entries with that name.
Update or drop the comment.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
Review Comment:
nit: "the earlier per-entry loop" is review history, master never had that
loop.
Same for "no longer" in SCOPE.md:132, I'd drop both.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
Review Comment:
We're the only broker, so a Java consumer sends all its partitions in one
ListOffsets.
Over 100 topics it gets INVALID_REQUEST, resends the exact same request and
never gets offsets.
Maybe answer the first 100 and give the rest a retriable code so each retry
makes progress?
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
Review Comment:
nit: stub path (handle_versioned_request) passes ERROR_UNSUPPORTED_VERSION,
this copy passes ERROR_INVALID_REQUEST.
Doesn't reach the wire today, but one shared version check + decode for both
paths avoids the drift.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
+///
+/// `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - Iggy
exposes no
+/// per-message timestamp index, so "offset of the first message at or after
timestamp T" for any
+/// other `T` cannot be answered without one. Those partitions get
[`ERROR_UNKNOWN_SERVER_ERROR`],
+/// not a fabricated offset.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+) -> Vec<ListOffsetsTopicResponse> {
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ for topic in requested {
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ partitions_by_name
+ .entry(topic.name.as_str())
+ .or_default()
+ .extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+
+ let mut watermarks_by_name: HashMap<&str, HighWatermarksResult> =
HashMap::new();
+ for (&name, partitions) in &partitions_by_name {
+ let result = if partitions.is_empty() {
+ Ok(Vec::new())
+ } else {
+ bridge.high_watermarks(name, partitions).await
+ };
+ watermarks_by_name.insert(name, result);
+ }
+
+ requested
+ .iter()
+ .map(|topic| {
+ // Always present: partitions_by_name (and so watermarks_by_name)
was built from
+ // exactly these same requested topic names, just above.
+ let watermarks = watermarks_by_name
+ .get(topic.name.as_str())
+ .expect("every requested topic name was resolved above");
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|requested| resolve_one_partition(requested, watermarks))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+ })
+ .collect()
+}
+
+/// One response entry for `topic`, every requested partition carrying
`error_code` - used for a
+/// whole-request failure (over the topic cap, or the aggregate deadline)
where no bridge call
+/// was made for this topic at all.
+fn error_topic_response(topic: &ListOffsetsTopic, error_code: i16) ->
ListOffsetsTopicResponse {
+ let partitions = topic
+ .partitions
+ .iter()
+ .map(|p| error_response(p.partition_index, error_code))
+ .collect();
+ ListOffsetsTopicResponse::default()
+ .with_name(topic.name.clone())
+ .with_partitions(partitions)
+}
+
+/// `watermarks` is the whole topic's batched result: `Err` is a call-level
failure (e.g. the
+/// mapped stream doesn't exist) applying to every partition alike; the inner
per-partition
+/// `Result` inside `Ok` is [`BridgeError::PartitionOutOfRange`] for one bad
index among otherwise
+/// resolvable ones.
+fn resolve_one_partition(
+ requested: &ListOffsetsPartition,
+ watermarks: &HighWatermarksResult,
+) -> 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 watermarks {
+ Err(call_err) => {
+ return error_response(requested.partition_index,
call_err.to_kafka_error_code());
+ }
+ Ok(results) => results,
+ };
Review Comment:
None of these call-level errors get logged, even though
bridge/error.rs:146-148 asks for it.
A rejected bridge login would show up as an error code on the client and
maybe one SDK log line.
I'd log once per topic in resolve_all_topics, debug! for client-caused ones
(unknown topic etc), error! otherwise.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
+ let topics = req
+ .topics
+ .iter()
+ .map(|topic| error_topic_response(topic, ERROR_INVALID_REQUEST))
+ .collect();
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"ListOffsets");
+ }
+
+ let topics =
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_all_topics(bridge, &req.topics)).await
+ {
+ Ok(topics) => topics,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "ListOffsets request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| error_topic_response(topic,
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ };
+ let resp = ListOffsetsResponse::default().with_topics(topics);
+ respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
+}
+
+/// Resolves every requested topic entry, deduping by name first so a topic
named in more than
+/// one request entry (or a name repeated verbatim) costs one
`high_watermarks` round trip, not
+/// one per entry - the earlier per-entry loop paid a separate round trip for
each, even for the
+/// same name.
+///
+/// `EARLIEST`/`LATEST` are the only timestamps this bridge resolves - Iggy
exposes no
+/// per-message timestamp index, so "offset of the first message at or after
timestamp T" for any
+/// other `T` cannot be answered without one. Those partitions get
[`ERROR_UNKNOWN_SERVER_ERROR`],
+/// not a fabricated offset.
+async fn resolve_all_topics(
+ bridge: &IggyBridge,
+ requested: &[ListOffsetsTopic],
+) -> Vec<ListOffsetsTopicResponse> {
+ let mut partitions_by_name: HashMap<&str, Vec<u32>> = HashMap::new();
+ for topic in requested {
+ let valid = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok());
+ partitions_by_name
+ .entry(topic.name.as_str())
+ .or_default()
+ .extend(valid);
+ }
+ for partitions in partitions_by_name.values_mut() {
+ partitions.sort_unstable();
+ partitions.dedup();
+ }
+
+ let mut watermarks_by_name: HashMap<&str, HighWatermarksResult> =
HashMap::new();
+ for (&name, partitions) in &partitions_by_name {
+ let result = if partitions.is_empty() {
+ Ok(Vec::new())
+ } else {
+ bridge.high_watermarks(name, partitions).await
Review Comment:
If all partitions of a topic ask for an unsupported timestamp we still pay
the whole high_watermarks round trip.
With the -1 retry loop that repeats on every retry.
Could check the timestamp before the lookup at 251-253 and skip the call
when nothing needs it.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
Review Comment:
nit: a big batch doesn't block other clients for its whole run.
The loop awaits one call at a time and the SDK lock is FIFO, so other
connections get in between.
Also the "Concurrency ceiling" section is in README.md, not
iggy_bridge/mod.rs.
##########
gateways/kafka/src/protocol/handlers/list_offsets.rs:
##########
@@ -17,44 +17,272 @@
//! `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 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_SERVER_ERROR,
ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ 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 may address through the
bridge.
+///
+/// `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
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"), so a large batch
head-of-line-blocks
+/// every other client's control plane for its duration. 100 keeps a
worst-case batch's aggregate
+/// bridge cost small relative to that shared resource while remaining
generous for any real
+/// consumer's offset lookup.
+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.
+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_INVALID_REQUEST)
+ });
+ }
+
+ 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 distinct_names: HashSet<&str> = req.topics.iter().map(|t|
t.name.as_str()).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "ListOffsets request addresses too many distinct topics; rejecting"
+ );
Review Comment:
nit: client controls this and the connection stays open, so a consumer stuck
retrying logs a warn every time.
debug! like the decode failure at 111-112?
--
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]
