ryerraguntla commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4077895327
##########
gateways/kafka/src/protocol/handlers/metadata.rs:
##########
@@ -53,23 +73,203 @@ 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) => topics.into_iter().map(found_result).collect(),
+ 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) => {
+ let distinct_names: HashSet<&str> =
names.iter().map(StrBytes::as_str).collect();
+ if distinct_names.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "Metadata request addresses too many distinct topics;
rejecting"
+ );
+ names
+ .iter()
+ .map(|name| error_result(name.clone(),
ERROR_INVALID_REQUEST))
+ .collect()
+ } else {
+ match tokio::time::timeout(REQUEST_DEADLINE,
resolve_named_topics(bridge, &names))
+ .await
+ {
+ Ok(results) => results,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_names.len(),
+ deadline_secs = REQUEST_DEADLINE.as_secs(),
+ "Metadata request's aggregate bridge work exceeded
its deadline; \
+ answering retriable instead of blocking further"
+ );
+ names
+ .iter()
+ .map(|name| error_result(name.clone(),
ERROR_REQUEST_TIMED_OUT))
+ .collect()
+ }
+ }
+ }
}
+ };
+
+ // `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, or one `list_kafka_topics()` call, can
still expand into a
+ // response `bounds_guard` never had the information to price in before
this bridge round
+ // trip returned. Checked here, before `encode_real_response` builds one
+ // `MetadataResponsePartition` per partition, not after - the expensive
part is building that
+ // `Vec`, not encoding the bytes that follow it.
+ 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",
+ )
+}
+
+/// 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
+}
+
+/// One resolved topic result for the real (bridge-backed) path -
`partitions_count` is
+/// meaningless when `error_code != ERROR_NONE`.
+struct TopicResult {
+ name: StrBytes,
+ error_code: i16,
+ partitions_count: u32,
+}
+
+fn found_result(metadata: crate::bridge::KafkaTopicMetadata) -> TopicResult {
+ TopicResult {
+ name: StrBytes::from_string(metadata.kafka_topic),
+ error_code: ERROR_NONE,
+ partitions_count: metadata.partitions_count,
+ }
+}
+
+async fn lookup_one_topic(bridge: &IggyBridge, name: StrBytes) -> TopicResult {
+ match bridge.get_kafka_topic(&name).await {
+ // `get_kafka_topic` (unlike `list_kafka_topics`) returns the SDK's
own `TopicDetails`,
+ // which carries the topic's raw Iggy-side name, not the Kafka-side
one under an override
+ // - so this echoes the caller's own `name`, not a field off the
result.
+ Ok(Some(details)) => TopicResult {
+ name,
+ error_code: ERROR_NONE,
+ partitions_count: details.partitions_count,
+ },
+ Ok(None) => TopicResult {
+ name,
+ error_code: ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ partitions_count: 0,
+ },
+ Err(err) => TopicResult {
+ name,
+ error_code: err.to_kafka_error_code(),
+ partitions_count: 0,
+ },
+ }
+}
+
+const fn error_result(name: StrBytes, error_code: i16) -> TopicResult {
+ TopicResult {
+ name,
+ error_code,
+ partitions_count: 0,
+ }
+}
+
+/// Resolves every requested name, deduping first so a name repeated in the
request (or asked
+/// about more than once, which the wire technically allows) costs one
`get_kafka_topic` round
+/// trip, not one per occurrence.
+async fn resolve_named_topics(bridge: &IggyBridge, names: &[StrBytes]) ->
Vec<TopicResult> {
+ let mut seen = HashSet::with_capacity(names.len());
+ let mut distinct = Vec::new();
+ for name in names {
+ if seen.insert(name.as_str()) {
+ distinct.push(name.clone());
+ }
+ }
+
+ let mut results_by_name: HashMap<&str, TopicResult> =
HashMap::with_capacity(distinct.len());
+ for name in &distinct {
+ let result = lookup_one_topic(bridge, name.clone()).await;
+ results_by_name.insert(name.as_str(), result);
+ }
+
+ names
+ .iter()
+ .map(|name| {
+ // Always present: `distinct` (and so results_by_name) was built
from exactly these
+ // same requested names, just above.
+ let cached = results_by_name
+ .get(name.as_str())
+ .expect("every requested name was resolved above");
Review Comment:
Metadata named-lookup response now dedupes to one entry per distinct name
(matching real Kafka's .toSet() behavior), not one per request occurrence.
resolve_named_topics simplified — no longer re-expands.
--
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]
