ryerraguntla commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4088468066
##########
gateways/kafka/src/protocol/handlers/metadata.rs:
##########
@@ -53,23 +73,249 @@ pub async fn handle(state: &GatewayState, api_version:
i16, body: Bytes) -> Hand
);
return HandleOutcome::Close;
}
- match decode_topics(api_version, body, state.max_frame_size) {
- Ok(topics) => respond_or_close(
- encode_response(api_version, &topics, &state.broker, ERROR_NONE),
- "Metadata",
- ),
+
+ let Some(bridge) = &state.bridge else {
+ return match decode_topics(api_version, body, state.max_frame_size) {
+ Ok(topics) => respond_or_close(
+ encode_response(api_version, &topics, &state.broker,
ERROR_NONE),
+ "Metadata",
+ ),
+ Err(error) => {
+ // Metadata has no top-level error field; a malformed body
cannot carry
+ // INVALID_REQUEST in a version-correct way for every client.
Close.
+ // debug!, not warn!: attacker-controlled, not
operator-actionable.
+ tracing::debug!(
+ %error,
+ api_version,
+ "Failed to decode Metadata request; closing connection"
+ );
+ HandleOutcome::Close
+ }
+ };
+ };
+
+ let requested = match decode_requested_topics(api_version, body,
state.max_frame_size) {
+ Ok(requested) => requested,
Err(error) => {
- // Metadata has no top-level error field; a malformed body cannot
carry
- // INVALID_REQUEST in a version-correct way for every client.
Close.
- // debug!, not warn!: attacker-controlled, not operator-actionable.
tracing::debug!(
%error,
api_version,
"Failed to decode Metadata request; closing connection"
);
- HandleOutcome::Close
+ return HandleOutcome::Close;
+ }
+ };
+
+ let results = match requested {
+ None => match bridge.list_kafka_topics().await {
+ Ok(topics) => {
+ let results: Vec<TopicResult> =
topics.into_iter().map(found_result).collect();
+ truncate_all_topics_to_frame_budget(results,
state.max_frame_size)
+ }
+ Err(error) => {
+ // Same "no top-level error field" constraint as a decode
failure: there is no
+ // way to answer "the bridge itself is unreachable" for an
all-topics request
+ // that doesn't also falsely claim zero topics exist.
+ tracing::warn!(%error, "Failed to list Kafka topics from the
Iggy bridge; closing connection");
+ return HandleOutcome::Close;
+ }
+ },
+ Some(names) => resolve_requested_named_topics(bridge, &names).await,
+ };
+
+ // `bounds_guard` cannot see this: it charges the projected response by
*requested* element
+ // (one topic name), but one real topic can carry up to Iggy's own
per-topic partition cap
+ // (1000) - a handful of names can still expand into a response
`bounds_guard` never had the
+ // information to price in before this bridge round trip returned. The
all-topics arm is
+ // pre-truncated to this same budget above
(`truncate_all_topics_to_frame_budget`) since its
+ // size is server-side, not client-controllable - closing over it would
take down every
+ // client's bootstrap Metadata call, permanently, the moment the catalog
grows past the
+ // trip point. This check is what still enforces the budget for the
named-lookup arm, where
+ // the cap (100 distinct topics) bounds the request but not what each one
costs to answer.
+ let total_partitions: usize = results
+ .iter()
+ .filter(|result| result.error_code == ERROR_NONE)
+ .map(|result| result.partitions_count as usize)
+ .sum();
+ if response_would_exceed_frame_size(total_partitions,
state.max_frame_size) {
+ tracing::warn!(
+ total_partitions,
+ max_frame_size = state.max_frame_size,
+ "Metadata response would exceed max_frame_size; closing connection"
+ );
+ return HandleOutcome::Close;
+ }
+
+ respond_or_close(
+ encode_real_response(api_version, &results, &state.broker),
+ "Metadata",
+ )
+}
+
+/// Resolves a named-lookup Metadata request's topics: dedupes first so every
path below (cap,
+/// deadline, success) builds its response from the distinct set rather than
one entry per
+/// request occurrence - real Kafka answers a topic named more than once with
one response entry,
+/// not one per repeat, and re-expanding to match the request let a handful of
repeats of one
+/// large topic name amplify a response sized off the repeat count instead of
the distinct count.
+async fn resolve_requested_named_topics(
+ bridge: &IggyBridge,
+ names: &[StrBytes],
+) -> Vec<TopicResult> {
+ let distinct = dedup_topic_names(names);
+ if distinct.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "Metadata request addresses too many distinct topics; rejecting"
+ );
+ return distinct
+ .iter()
+ .map(|name| error_result(name.clone(), ERROR_INVALID_REQUEST))
+ .collect();
+ }
+
+ match tokio::time::timeout(REQUEST_DEADLINE, resolve_named_topics(bridge,
&distinct)).await {
+ Ok(results) => results,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "Metadata request's aggregate bridge work exceeded its
deadline; answering \
+ retriable instead of blocking further"
+ );
+ distinct
+ .iter()
+ .map(|name| error_result(name.clone(),
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ }
+}
+
+/// Conservative per-partition byte cost of one encoded
`MetadataResponsePartition` at v9 (the
+/// densest wire shape this handler emits): measured ~26 bytes (`error_code` +
`partition_index` +
+/// `leader_id` + `leader_epoch` + a 1-entry `replica_nodes` + a 1-entry
`isr_nodes` + an empty
+/// `offline_replicas` + tagged fields). 64 matches the margin
`bounds_guard`'s own
+/// `RESPONSE_BYTES_PER_ELEMENT` uses for the same kind of estimate, rather
than shaving this to
+/// the measured minimum.
+const RESPONSE_BYTES_PER_PARTITION: usize = 64;
+
+const fn response_would_exceed_frame_size(total_partitions: usize,
max_frame_size: usize) -> bool {
+ total_partitions.saturating_mul(RESPONSE_BYTES_PER_PARTITION) >
max_frame_size
+}
+
+/// Trims an all-topics [`IggyBridge::list_kafka_topics`] result to fit
`max_frame_size`, keeping
+/// as many whole topics (in listing order) as the budget allows.
+///
+/// Unlike the named-lookup arm, the all-topics response's size is a
server-side property (the
+/// cluster's total partition count) that the requesting client never chose
and cannot shrink -
+/// closing the connection over it, as the shared frame-size guard below does
for the
+/// client-controllable named-lookup case, would make every all-topics
Metadata call fail
+/// identically and permanently once the catalog crosses the trip point. That
call is the
+/// bootstrap and refresh shape both librdkafka and the Java client use, so a
hard close reads as
+/// "broker down" and reconnect-loops rather than surfacing a usable, if
partial, result. This
+/// wire protocol has no pagination cursor to ask for the rest with, so a
truncated list - honest
+/// about being incomplete via the dropped entries, not via an error code - is
what's available.
+fn truncate_all_topics_to_frame_budget(
+ mut results: Vec<TopicResult>,
+ max_frame_size: usize,
+) -> Vec<TopicResult> {
+ let mut cumulative_partitions = 0usize;
+ let mut keep = results.len();
+ for (index, result) in results.iter().enumerate() {
+ if result.error_code != ERROR_NONE {
+ continue;
+ }
+ let next = cumulative_partitions + result.partitions_count as usize;
Review Comment:
New estimated_response_bytes() charges fixed per-topic overhead (13 bytes) +
actual name length + partition cost — and charges it for every result, not
just ERROR_NONE ones (error results still encode a full wrapper with an empty
partitions array). Used by both the named-lookup guard and the all-topics
truncation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
