ryerraguntla commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4077908338
##########
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:
error_message no longer re-embeds the topic name via err.to_string() (which
InvalidKafkaTopicName/InvalidPartitionCount's Display does) — new
error_message_for extracts just the reason text.
--
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]
