You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "ryerraguntla (via GitHub)" <gi...@apache.org> on 2026/09/22 03:09:03 UTC
[PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
ryerraguntla opened a new pull request, #4259:
URL: https://github.com/apache/iggy/pull/4259
## Summary
Wires the Kafka gateway's ListOffsets handler to the real Iggy bridge instead of the stub added in #3421. Standalone branch, no file overlap with the combined CreateTopics+Metadata PR — safe to review/merge in
parallel.
With IGGY_KAFKA_BRIDGE_ENABLED=true:
- LATEST answers from IggyBridge::high_watermarks.
- EARLIEST answers 0.
- Any other requested timestamp (arbitrary-timestamp offset search) is unsupported — Iggy exposes no per-message timestamp index — and answers UNKNOWN_SERVER_ERROR (-1) per partition rather than a fabricated
offset.
With the bridge off (default), behavior is unchanged from #3421: NOT_LEADER_OR_FOLLOWER (6).
Closes #3537.
## What's included
- high_watermarks-backed LATEST/EARLIEST resolution, one bridge round trip per distinct topic.
- Bridge fan-out hardening, consistent with the CreateTopics/Metadata PR: dedup requested topic names before any bridge call, MAX_BRIDGE_BACKED_TOPICS = 100 cap (INVALID_REQUEST, no bridge call above it), fixed
20s wall-clock deadline over the request's aggregate bridge work (no timeout_ms field exists in any version this gateway supports, v0–v6).
- New error code ERROR_REQUEST_TIMED_OUT used on deadline expiry.
## Known gap (not fixed here, tracked separately)
EARLIEST = 0 is correct only for a partition that has never had retention trim its first segment. Iggy currently tracks no rolling low-watermark distinct from partition creation, so once retention purges a
partition's first segment, 0 names a log-start offset that no longer exists — a real consumer with auto.offset.reset=earliest would seek into a hole. Not fixable client-side; needs a core-server change
(separately drafted as a GitHub issue, not yet filed). Harmless today only because Fetch (#3536) is still a stub — nothing yet reads at the offset this returns.
Test plan
- cargo build -p iggy-gateway-kafka --all-targets — clean.
- cargo clippy -p iggy-gateway-kafka --all-features --all-targets -- -D warnings — no issues.
- cargo test -p iggy-gateway-kafka — 291 passed, 15 suites (includes new tests/list_offsets_real_bridge_tests.rs, 332 lines).
## AI Usage
If AI tools were used, please answer:
1. Which tools? Claude code for implementation , Cursor for review and validation
2. Scope of usage? Implementation , review and validation
3. How did you verify the generated code works correctly? Test Suite and Code walkthrough
4. Can you explain every line of the code if asked? Yes
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4076566104
##########
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:
Deadline now applied per-topic via timeout_at, not once around the whole batch. Already resolved topics keep their answer; only not-yet-started topics get REQUEST_TIMED_OUT
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077014793
##########
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:
Downgraded to debug!, consistent with the decode-failure precedent already in the file
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4259:
URL: https://github.com/apache/iggy/pull/4259#issuecomment-5787053802
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077011370
##########
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:
Downgraded to debug!, consistent with the decode-failure precedent already in the file.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex merged PR #4259:
URL: https://github.com/apache/iggy/pull/4259
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4076984599
##########
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:
Replaced with binary_search_by_key; added a test with a multi-entry sorted result to exercise it (single-entry would pass a broken binary search too).
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077011370
##########
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:
Dropped from both the doc comment and SCOPE.md:132.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4076997438
##########
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:
New TopicLookup::NoLookupNeeded variant — call skipped entirely when no partition in a topic asks LATEST/EARLIEST.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4076588954
##########
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:
Cap now resolves the first 100 (request order) and answers the rest REQUEST_TIMED_OUT (retriable) with no bridge call — a client retrying only its erroring topics narrows below the cap within a couple of retries
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077004526
##########
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:
Rewritten to describe the real binary-search invariant.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077318037
##########
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:
Nonexistent topics now correctly report UNKNOWN_TOPIC_OR_PARTITION regardless of what timestamp was asked.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077007234
##########
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:
Fixed file reference; softened claim to "blocks only the call in flight."
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077001296
##########
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:
Now sends ERROR_UNSUPPORTED_VERSION, matching the stub path.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077324745
##########
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:
New integration test distinct_partitions_resolve_to_their_own_distinct_watermarks — partition 0 seeded with 3 messages, partition 1 with 7, requested out of index order, asserts each gets its own watermark. Pins the binary search against a broken-search/constant-index regression the old two-empty-partitions test couldn't catch.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4259:
URL: https://github.com/apache/iggy/pull/4259#issuecomment-5770733465
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4259?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 26.55%. Comparing base ([`7bd6967`](https://app.codecov.io/gh/apache/iggy/commit/7bd69670f9ad9946f6490359443ec25bf7e2a089?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`03803c7`](https://app.codecov.io/gh/apache/iggy/commit/03803c72144129dcd2eb88dbd76c7edde943b1b6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4259 +/- ##
=============================================
- Coverage 87.54% 26.55% -61.00%
Complexity 1575 1575
=============================================
Files 1284 1281 -3
Lines 223942 181110 -42832
Branches 187305 144473 -42832
=============================================
- Hits 196050 48093 -147957
- Misses 23168 131322 +108154
+ Partials 4724 1695 -3029
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `11.15% <33.79%> (-77.48%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.68% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.41% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.97% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.67% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.36% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4259/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.05% <ø> (ø)` | |
[see 688 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4259/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077329408
##########
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:
Fixed by the same revert as earlier one — needs_lookup (the thing whose insert bypassed the index guard) no longer exists. group_requested_topics now only ever pushes valid indices, full stop. │
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077332072
##########
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:
.config/nextest.toml — added list_offsets_real_bridge_tests to the kafka_bridge group filter. Verified via cargo nextest show-config test-groups: both binaries now correctly grouped under max-threads = 4
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4077321917
##########
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:
README.md + iggy_bridge/mod.rs doc updated — dropped the false "nothing calls this bridge from a live handler yet" claim, now correctly states ListOffsets is live and flags the missing concurrency bound as still-open, more pressing once CreateTopics/Metadata/Produce/Fetch land.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): Implement ListOffsets Kafka Keys backed by Iggy bridge [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4259:
URL: https://github.com/apache/iggy/pull/4259#discussion_r4076977102
##########
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:
Logged once per topic in the lookup loop: debug! for client-caused (UNKNOWN_TOPIC_OR_PARTITION), error! otherwise.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org