numinnex commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4072774772
##########
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;
Review Comment:
This guard also covers the all-topics arm, where `total_partitions` is the
whole catalog rather than anything the client asked for. With `max_frame_size`
defaulting to 8 MiB the trip point is 131,072 partitions cluster-wide, after
which every all-topics Metadata is a bare TCP close with no error code, for
every client, permanently. All-topics is the bootstrap and refresh shape for
librdkafka and the Java client, which read a close as "broker down" and
reconnect-loop. Truncating seems better than closing here.
##########
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:
The bridge calls are deduped, but this map runs over every *occurrence*, so
the response carries one entry per repeat. Real Kafka does
`metadataRequest.topics.asScala.toSet` and answers once per distinct name. At
v9, 131 copies of one 1000-partition name passes `bounds_guard` (131 elements),
passes the 100-distinct cap (1 distinct), costs one round trip, and clears the
new frame guard at 8,384,000 bytes, while building 131,000 partition records
from a ~1 KB request.
##########
gateways/kafka/src/protocol/handlers/create_topics.rs:
##########
@@ -17,44 +17,324 @@
//! `CreateTopics` (API key 19).
+use std::collections::HashSet;
+use std::time::Duration;
+
use bytes::Bytes;
use kafka_protocol::messages::create_topics_request::CreatableTopic;
use kafka_protocol::messages::create_topics_response::CreatableTopicResult;
-use kafka_protocol::messages::{CreateTopicsRequest, CreateTopicsResponse};
+use kafka_protocol::messages::{CreateTopicsRequest, CreateTopicsResponse,
TopicName};
+use kafka_protocol::protocol::StrBytes;
+use crate::bridge::{IggyBridge, TopicCreationOutcome};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_PARTITIONS,
- ERROR_INVALID_REPLICATION_FACTOR, ERROR_NONE, ERROR_NOT_CONTROLLER,
GatewayState,
- HandleOutcome,
+ API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_CONFIG,
ERROR_INVALID_PARTITIONS,
+ ERROR_INVALID_REPLICATION_FACTOR, ERROR_INVALID_REQUEST, ERROR_NONE,
ERROR_NOT_CONTROLLER,
+ ERROR_REQUEST_TIMED_OUT, ERROR_TOPIC_ALREADY_EXISTS, GatewayState,
HandleOutcome,
};
use crate::protocol::bounds_guard::validate_create_topics_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_CREATE_TOPICS,
min_version: 2,
max_version: 5,
};
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until a handler awaits the
bridge"
-)]
+/// KIP-464 `num_partitions = -1` with no manual assignment: the count this
bridge creates.
+///
+/// Matches real Kafka's own out-of-box `num.partitions=1` broker default. Not
read from any
+/// bridge config - there is no such config surface today.
+const DEFAULT_PARTITION_COUNT: u32 = 1;
+
+/// Cap on distinct topic names one `CreateTopics` 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 non-duplicate requested name here costs up to ~4 Iggy
round trips
+/// (`ensure_stream` + `create_topic`, plus a possible race-retry read on
either) against the
+/// single lockstep `IggyClient` every Kafka connection on this gateway shares
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"). 100 keeps a
worst-case batch's
+/// aggregate bridge cost small relative to that shared resource while
remaining generous for any
+/// real admin batch. Duplicate names never count against this cap - they're
rejected by
+/// [`find_duplicate_names`] before ever reaching the bridge.
+const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
+
+/// Bounds imposed on the request's own `timeout_ms` before it becomes the
aggregate bridge-work
+/// deadline. That value is client-supplied and otherwise unchecked: `0` or
negative would abort
+/// every topic on arrival, and an oversized one would tie up the shared
`IggyClient` past any
+/// reasonable request.
+const MIN_REQUEST_TIMEOUT: Duration = Duration::from_millis(1_000);
+const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Clamps the wire's own `timeout_ms` (KIP-4's field for exactly this) into
+/// `[MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT]` - unlike
`ListOffsets`/`Metadata`, `CreateTopics`
+/// carries a real client-supplied deadline to honor, not just a fixed
internal ceiling.
+fn clamp_request_timeout(timeout_ms: i32) -> Duration {
+ let requested =
Duration::from_millis(u64::try_from(timeout_ms).unwrap_or(0));
+ requested.clamp(MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT)
+}
+
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) ->
HandleOutcome {
- handle_versioned_request(
- API_KEY_CREATE_TOPICS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
- validate_create_topics_shape(v, b, state.max_frame_size)
+ let Some(bridge) = &state.bridge else {
+ return handle_versioned_request(
+ API_KEY_CREATE_TOPICS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
+ validate_create_topics_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "CreateTopics",
+ );
+ };
+
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |version| {
+ encode_error_response(version, ERROR_INVALID_REQUEST)
+ });
+ }
+
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_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 CreateTopics request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let duplicate_names = find_duplicate_names(&req.topics);
+
+ let distinct_bridge_backed: HashSet<&TopicName> = req
+ .topics
+ .iter()
+ .map(|topic| &topic.name)
+ .filter(|name| !duplicate_names.contains(*name))
+ .collect();
+ if distinct_bridge_backed.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_bridge_backed.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "CreateTopics request addresses too many distinct topics;
rejecting"
+ );
+ let results = req
+ .topics
+ .iter()
+ .map(|topic| {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_INVALID_REQUEST)
})
- },
- encode_response,
- encode_error_response,
- "CreateTopics",
+ .collect();
+ let resp = CreateTopicsResponse::default().with_topics(results);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"CreateTopics");
+ }
+
+ let deadline = clamp_request_timeout(req.timeout_ms);
+ let results = match tokio::time::timeout(
+ deadline,
+ create_all_topics(
+ bridge,
+ api_version,
+ &req.topics,
+ &duplicate_names,
+ req.validate_only,
+ ),
)
+ .await
+ {
+ Ok(results) => results,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_bridge_backed.len(),
+ deadline_ms = deadline.as_millis(),
+ "CreateTopics request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ })
+ .collect()
+ }
+ };
+ let resp = CreateTopicsResponse::default().with_topics(results);
+ respond_or_close(encode_message(&resp, api_version, 256), "CreateTopics")
+}
+
+/// Creates (or reports on) every requested topic, skipping the bridge
entirely for a duplicate
+/// name - real Kafka refuses the whole name, not a first-wins/last-wins
split: creating one
+/// occurrence and reporting `TOPIC_ALREADY_EXISTS` for the other would let a
client observe a
+/// create it never got a `NONE` for (`AdminClient` keys its futures by name,
so a second
+/// per-topic result for the same name is silently discarded client-side
regardless of which one
+/// this bridge picked).
+async fn create_all_topics(
+ bridge: &IggyBridge,
+ api_version: i16,
+ topics: &[CreatableTopic],
+ duplicate_names: &HashSet<TopicName>,
+ validate_only: bool,
+) -> Vec<CreatableTopicResult> {
+ let mut results = Vec::with_capacity(topics.len());
+ for topic in topics {
+ let result = if duplicate_names.contains(&topic.name) {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_INVALID_REQUEST)
+ } else {
+ create_one_topic(bridge, api_version, topic, validate_only).await
+ };
+ results.push(result);
+ }
+ results
+}
+
+/// Every topic name that appears more than once in `topics` - real Kafka
+/// (`ControllerApis.createTopics`) refuses every occurrence of a duplicate
name with
+/// `INVALID_REQUEST` (42) and creates nothing for it, rather than creating
the first occurrence
+/// and reporting the rest as already existing.
+fn find_duplicate_names(topics: &[CreatableTopic]) -> HashSet<TopicName> {
+ let mut seen = HashSet::with_capacity(topics.len());
+ let mut duplicates = HashSet::new();
+ for topic in topics {
+ if !seen.insert(topic.name.clone()) {
+ duplicates.insert(topic.name.clone());
+ }
+ }
+ duplicates
+}
+
+/// Validates and, when the topic is not rejected outright, provisions one
requested topic.
+///
+/// `configs` is rejected before the shape check - a config-bearing request is
rejected the same
+/// way regardless of how its partitions/replication are shaped.
+///
+/// `validate_only` and the real create path diverge deliberately below that
point, not just in
+/// whether they call the bridge: `validate_only` never mutates anything, so a
plain existence
+/// read ([`IggyBridge::get_kafka_topic`]) is fine - there's no race to
protect against when
+/// nothing gets created either way. The real path instead calls
+/// [`IggyBridge::create_kafka_topic`], which folds the existence check and
the create into one
+/// atomic call - a separate read-then-write here would let two concurrent
`CreateTopics` for the
+/// same new name both observe `Ok(None)` and both receive `NONE`, when Kafka
guarantees exactly
+/// one caller does.
+async fn create_one_topic(
+ bridge: &IggyBridge,
+ version: i16,
+ topic: &CreatableTopic,
+ validate_only: bool,
+) -> CreatableTopicResult {
+ let result = CreatableTopicResult::default().with_name(topic.name.clone());
+
+ if !topic.configs.is_empty() {
+ return result
+ .with_error_code(ERROR_INVALID_CONFIG)
+ .with_error_message(Some(StrBytes::from(
+ "per-topic configs are not supported by this
bridge".to_string(),
+ )));
+ }
+
+ let partition_count = match validate_create_topic_shape(version, topic) {
+ Ok(count) => count,
+ Err(code) => return result.with_error_code(code),
+ };
+
+ let kafka_topic = topic.name.as_str();
+ let success = || {
+ result
+ .clone()
+ .with_error_code(ERROR_NONE)
+
.with_num_partitions(i32::try_from(partition_count).unwrap_or(i32::MAX))
+ .with_replication_factor(1)
+ };
+
+ if validate_only {
+ return match bridge.get_kafka_topic(kafka_topic).await {
+ Ok(Some(_existing)) =>
result.with_error_code(ERROR_TOPIC_ALREADY_EXISTS),
+ Ok(None) => success(),
+ Err(err) => result
+ .with_error_code(err.to_kafka_error_code())
+ .with_error_message(Some(StrBytes::from(err.to_string()))),
+ };
+ }
+
+ match bridge
+ .create_kafka_topic(kafka_topic, partition_count)
+ .await
+ {
+ Ok(TopicCreationOutcome::Created) => success(),
+ Ok(TopicCreationOutcome::AlreadyExists) => {
+ result.with_error_code(ERROR_TOPIC_ALREADY_EXISTS)
+ }
+ Err(err) => result
+ .with_error_code(err.to_kafka_error_code())
+ .with_error_message(Some(StrBytes::from(err.to_string()))),
+ }
+}
+
+/// Validates one requested topic's KIP-464 shape and resolves its partition
count.
+///
+/// A manual partition `assignments` list and an explicit
`num_partitions`/`replication_factor`
+/// are mutually exclusive inputs, not two independently-checked values that
happen to agree: real
+/// Kafka's own `ReplicationControlManager` rejects a manual assignment unless
both are exactly
+/// `-1`, regardless of whether an explicit `num_partitions` matches
`assignments.len()`. A count
+/// that only *disagrees* with the assignment length is not a distinct, more
lenient case - the
+/// combination itself is what's invalid, so both are `INVALID_REQUEST` (42),
never `NONE`.
+///
+/// With no assignments, `num_partitions = -1` / `replication_factor = -1`
mean "use the broker
+/// default" from v4+ (pre-v4 requires an explicit positive value for both,
since v2/v3 have no
+/// broker-default sentinel absent a manual assignment).
+fn validate_create_topic_shape(
+ version: i16,
+ topic: &CreatableTopic,
+) -> core::result::Result<u32, i16> {
+ if !topic.assignments.is_empty() {
+ if topic.num_partitions != -1 || topic.replication_factor != -1 {
+ return Err(ERROR_INVALID_REQUEST);
+ }
+ return
Ok(u32::try_from(topic.assignments.len()).unwrap_or(DEFAULT_PARTITION_COUNT));
Review Comment:
`assignments.len()` is the only thing read here, so the partition indices
themselves are never checked. Kafka's `ReplicationControlManager.createTopic`
answers `INVALID_REPLICA_ASSIGNMENT` (39) both for a duplicate
`partition_index` and for a sequence that isn't consecutive from 0. `new
NewTopic(name, Map.of(5, List.of(1), 7, List.of(1)))` sends `-1/-1`, clears the
gate above, and gets `NONE` with `num_partitions=2` for a topic whose
partitions are 0 and 1. 39 is now absent from `src/` entirely, so there's no
code left for the conditions Kafka does use it for.
##########
gateways/kafka/src/bridge/error.rs:
##########
@@ -163,7 +172,17 @@ const fn iggy_error_to_kafka_code(err: &IggyError) -> i16 {
| IggyError::TcpError
| IggyError::TransientNotAccepted => ERROR_NOT_LEADER_OR_FOLLOWER,
IggyError::TransientNotCommitted => ERROR_REQUEST_TIMED_OUT,
- IggyError::TooManyPartitions => ERROR_INVALID_PARTITIONS,
+ // Not `ERROR_INVALID_PARTITIONS` (37): that code's own text, per
`kafka-protocol`'s
+ // table, is "Number of partitions is below 1" - the opposite
condition from "too many"
+ // (Iggy's server-side cap, above 1000). Reusing 37 for both
directions would return a
+ // client-visible error message that contradicts the actual request it
sent.
+ IggyError::TooManyPartitions => ERROR_INVALID_REQUEST,
+ // The operation *did* commit - the SDK's own reconnect path replayed
a write whose first
+ // attempt already applied, and the server's client-table dedup caught
the replay. Falling
+ // into the catch-all below would report a permanent server fault for
a request that
+ // actually succeeded; a Java client treats `UNKNOWN_SERVER_ERROR` as
non-retriable and
+ // would surface a spurious failure for a `CreateTopics` that in fact
created the topic.
+ IggyError::RequestAlreadyApplied => ERROR_NONE,
Review Comment:
The reasoning above is entirely about CreateTopics, but this arm is global.
Reached through `metadata.rs`, it produces `TopicResult { error_code: NONE,
partitions_count: 0 }`, and `encode_real_response` then emits a topic with
error 0 and an empty partitions array: the topic appears to exist with nowhere
to produce, and no retriable error for the client to act on. Scoping this to
the CreateTopics write path keeps the fix without that side effect.
##########
gateways/kafka/src/protocol/handlers/create_topics.rs:
##########
@@ -17,44 +17,324 @@
//! `CreateTopics` (API key 19).
+use std::collections::HashSet;
+use std::time::Duration;
+
use bytes::Bytes;
use kafka_protocol::messages::create_topics_request::CreatableTopic;
use kafka_protocol::messages::create_topics_response::CreatableTopicResult;
-use kafka_protocol::messages::{CreateTopicsRequest, CreateTopicsResponse};
+use kafka_protocol::messages::{CreateTopicsRequest, CreateTopicsResponse,
TopicName};
+use kafka_protocol::protocol::StrBytes;
+use crate::bridge::{IggyBridge, TopicCreationOutcome};
use crate::error::Result;
use crate::protocol::api::{
- API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_PARTITIONS,
- ERROR_INVALID_REPLICATION_FACTOR, ERROR_NONE, ERROR_NOT_CONTROLLER,
GatewayState,
- HandleOutcome,
+ API_KEY_CREATE_TOPICS, ApiVersionRange, ERROR_INVALID_CONFIG,
ERROR_INVALID_PARTITIONS,
+ ERROR_INVALID_REPLICATION_FACTOR, ERROR_INVALID_REQUEST, ERROR_NONE,
ERROR_NOT_CONTROLLER,
+ ERROR_REQUEST_TIMED_OUT, ERROR_TOPIC_ALREADY_EXISTS, GatewayState,
HandleOutcome,
};
use crate::protocol::bounds_guard::validate_create_topics_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_CREATE_TOPICS,
min_version: 2,
max_version: 5,
};
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until a handler awaits the
bridge"
-)]
+/// KIP-464 `num_partitions = -1` with no manual assignment: the count this
bridge creates.
+///
+/// Matches real Kafka's own out-of-box `num.partitions=1` broker default. Not
read from any
+/// bridge config - there is no such config surface today.
+const DEFAULT_PARTITION_COUNT: u32 = 1;
+
+/// Cap on distinct topic names one `CreateTopics` 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 non-duplicate requested name here costs up to ~4 Iggy
round trips
+/// (`ensure_stream` + `create_topic`, plus a possible race-retry read on
either) against the
+/// single lockstep `IggyClient` every Kafka connection on this gateway shares
+/// (`bridge/iggy_bridge/mod.rs`'s "Concurrency ceiling"). 100 keeps a
worst-case batch's
+/// aggregate bridge cost small relative to that shared resource while
remaining generous for any
+/// real admin batch. Duplicate names never count against this cap - they're
rejected by
+/// [`find_duplicate_names`] before ever reaching the bridge.
+const MAX_BRIDGE_BACKED_TOPICS: usize = 100;
+
+/// Bounds imposed on the request's own `timeout_ms` before it becomes the
aggregate bridge-work
+/// deadline. That value is client-supplied and otherwise unchecked: `0` or
negative would abort
+/// every topic on arrival, and an oversized one would tie up the shared
`IggyClient` past any
+/// reasonable request.
+const MIN_REQUEST_TIMEOUT: Duration = Duration::from_millis(1_000);
+const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Clamps the wire's own `timeout_ms` (KIP-4's field for exactly this) into
+/// `[MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT]` - unlike
`ListOffsets`/`Metadata`, `CreateTopics`
+/// carries a real client-supplied deadline to honor, not just a fixed
internal ceiling.
+fn clamp_request_timeout(timeout_ms: i32) -> Duration {
+ let requested =
Duration::from_millis(u64::try_from(timeout_ms).unwrap_or(0));
+ requested.clamp(MIN_REQUEST_TIMEOUT, MAX_REQUEST_TIMEOUT)
+}
+
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) ->
HandleOutcome {
- handle_versioned_request(
- API_KEY_CREATE_TOPICS,
- api_version,
- body,
- |v, b| {
- decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
- validate_create_topics_shape(v, b, state.max_frame_size)
+ let Some(bridge) = &state.bridge else {
+ return handle_versioned_request(
+ API_KEY_CREATE_TOPICS,
+ api_version,
+ body,
+ |v, b| {
+ decode_guarded::<CreateTopicsRequest>(v, b, |v, b| {
+ validate_create_topics_shape(v, b, state.max_frame_size)
+ })
+ },
+ encode_response,
+ encode_error_response,
+ "CreateTopics",
+ );
+ };
+
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |version| {
+ encode_error_response(version, ERROR_INVALID_REQUEST)
+ });
+ }
+
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_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 CreateTopics request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let duplicate_names = find_duplicate_names(&req.topics);
+
+ let distinct_bridge_backed: HashSet<&TopicName> = req
+ .topics
+ .iter()
+ .map(|topic| &topic.name)
+ .filter(|name| !duplicate_names.contains(*name))
+ .collect();
+ if distinct_bridge_backed.len() > MAX_BRIDGE_BACKED_TOPICS {
+ tracing::warn!(
+ distinct_topics = distinct_bridge_backed.len(),
+ max = MAX_BRIDGE_BACKED_TOPICS,
+ "CreateTopics request addresses too many distinct topics;
rejecting"
+ );
+ let results = req
+ .topics
+ .iter()
+ .map(|topic| {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_INVALID_REQUEST)
})
- },
- encode_response,
- encode_error_response,
- "CreateTopics",
+ .collect();
+ let resp = CreateTopicsResponse::default().with_topics(results);
+ return respond_or_close(encode_message(&resp, api_version, 256),
"CreateTopics");
+ }
+
+ let deadline = clamp_request_timeout(req.timeout_ms);
+ let results = match tokio::time::timeout(
+ deadline,
+ create_all_topics(
+ bridge,
+ api_version,
+ &req.topics,
+ &duplicate_names,
+ req.validate_only,
+ ),
)
+ .await
+ {
+ Ok(results) => results,
+ Err(_elapsed) => {
+ tracing::warn!(
+ distinct_topics = distinct_bridge_backed.len(),
+ deadline_ms = deadline.as_millis(),
+ "CreateTopics request's aggregate bridge work exceeded its
deadline; \
+ answering retriable instead of blocking further"
+ );
+ req.topics
+ .iter()
+ .map(|topic| {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_REQUEST_TIMED_OUT)
+ })
+ .collect()
+ }
+ };
+ let resp = CreateTopicsResponse::default().with_topics(results);
+ respond_or_close(encode_message(&resp, api_version, 256), "CreateTopics")
+}
+
+/// Creates (or reports on) every requested topic, skipping the bridge
entirely for a duplicate
+/// name - real Kafka refuses the whole name, not a first-wins/last-wins
split: creating one
+/// occurrence and reporting `TOPIC_ALREADY_EXISTS` for the other would let a
client observe a
+/// create it never got a `NONE` for (`AdminClient` keys its futures by name,
so a second
+/// per-topic result for the same name is silently discarded client-side
regardless of which one
+/// this bridge picked).
+async fn create_all_topics(
+ bridge: &IggyBridge,
+ api_version: i16,
+ topics: &[CreatableTopic],
+ duplicate_names: &HashSet<TopicName>,
+ validate_only: bool,
+) -> Vec<CreatableTopicResult> {
+ let mut results = Vec::with_capacity(topics.len());
+ for topic in topics {
+ let result = if duplicate_names.contains(&topic.name) {
+ CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_INVALID_REQUEST)
+ } else {
+ create_one_topic(bridge, api_version, topic, validate_only).await
+ };
+ results.push(result);
+ }
+ results
+}
+
+/// Every topic name that appears more than once in `topics` - real Kafka
+/// (`ControllerApis.createTopics`) refuses every occurrence of a duplicate
name with
+/// `INVALID_REQUEST` (42) and creates nothing for it, rather than creating
the first occurrence
+/// and reporting the rest as already existing.
+fn find_duplicate_names(topics: &[CreatableTopic]) -> HashSet<TopicName> {
+ let mut seen = HashSet::with_capacity(topics.len());
+ let mut duplicates = HashSet::new();
+ for topic in topics {
+ if !seen.insert(topic.name.clone()) {
+ duplicates.insert(topic.name.clone());
+ }
+ }
+ duplicates
+}
+
+/// Validates and, when the topic is not rejected outright, provisions one
requested topic.
+///
+/// `configs` is rejected before the shape check - a config-bearing request is
rejected the same
+/// way regardless of how its partitions/replication are shaped.
+///
+/// `validate_only` and the real create path diverge deliberately below that
point, not just in
+/// whether they call the bridge: `validate_only` never mutates anything, so a
plain existence
+/// read ([`IggyBridge::get_kafka_topic`]) is fine - there's no race to
protect against when
+/// nothing gets created either way. The real path instead calls
+/// [`IggyBridge::create_kafka_topic`], which folds the existence check and
the create into one
+/// atomic call - a separate read-then-write here would let two concurrent
`CreateTopics` for the
+/// same new name both observe `Ok(None)` and both receive `NONE`, when Kafka
guarantees exactly
+/// one caller does.
+async fn create_one_topic(
+ bridge: &IggyBridge,
+ version: i16,
+ topic: &CreatableTopic,
+ validate_only: bool,
+) -> CreatableTopicResult {
+ let result = CreatableTopicResult::default().with_name(topic.name.clone());
+
+ if !topic.configs.is_empty() {
+ return result
+ .with_error_code(ERROR_INVALID_CONFIG)
+ .with_error_message(Some(StrBytes::from(
+ "per-topic configs are not supported by this
bridge".to_string(),
+ )));
+ }
+
+ let partition_count = match validate_create_topic_shape(version, topic) {
+ Ok(count) => count,
+ Err(code) => return result.with_error_code(code),
+ };
+
+ let kafka_topic = topic.name.as_str();
+ let success = || {
+ result
+ .clone()
+ .with_error_code(ERROR_NONE)
+
.with_num_partitions(i32::try_from(partition_count).unwrap_or(i32::MAX))
+ .with_replication_factor(1)
+ };
+
+ if validate_only {
+ return match bridge.get_kafka_topic(kafka_topic).await {
+ Ok(Some(_existing)) =>
result.with_error_code(ERROR_TOPIC_ALREADY_EXISTS),
+ Ok(None) => success(),
+ Err(err) => result
+ .with_error_code(err.to_kafka_error_code())
+ .with_error_message(Some(StrBytes::from(err.to_string()))),
+ };
+ }
+
+ match bridge
+ .create_kafka_topic(kafka_topic, partition_count)
+ .await
+ {
+ Ok(TopicCreationOutcome::Created) => success(),
+ Ok(TopicCreationOutcome::AlreadyExists) => {
+ result.with_error_code(ERROR_TOPIC_ALREADY_EXISTS)
+ }
+ Err(err) => result
+ .with_error_code(err.to_kafka_error_code())
+ .with_error_message(Some(StrBytes::from(err.to_string()))),
Review Comment:
`err.to_string()` embeds the full wire name for `InvalidKafkaTopicName`, so
each result echoes the name twice while `bounds_guard` charged it once: the
same one-echo modelling error this PR fixes on the Metadata side.
`create_kafka_topic` validates the name before any I/O, so 100 names of 64 KiB
clear the guard, the cap and the frame check, cost zero bridge calls, and build
a ~13 MB response that `send_response` admits because it still only checks
`i32::MAX`.
--
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]
