numinnex commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4082045181
##########
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:
This charges only `partitions × 64`, so topic names, per-topic overhead and
zero-partition topics are free. Measured: 500 one-partition topics with
255-byte names encode to 146,037 B against 32,000 charged, and 200,000
zero-partition topics give a ~53 MB response against the 8 MiB budget.
`send_response` only checks `i32::MAX`, so nothing downstream catches it, and
the one-partition case can be built through CreateTopics without
authentication. Charging each topic's real size (name, fixed overhead,
partitions), or encoding and then measuring, would close it.
##########
gateways/kafka/src/protocol/handlers/create_topics.rs:
##########
@@ -17,44 +17,378 @@
//! `CreateTopics` (API key 19).
+use std::collections::HashSet;
+use std::time::Duration;
+
use bytes::Bytes;
-use kafka_protocol::messages::create_topics_request::CreatableTopic;
+use iggy::prelude::IggyError;
+use
kafka_protocol::messages::create_topics_request::{CreatableReplicaAssignment,
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::{BridgeError, 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_REPLICA_ASSIGNMENT, 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(error_message_for(&err)))),
+ };
+ }
+
+ match bridge
+ .create_kafka_topic(kafka_topic, partition_count)
+ .await
+ {
+ // The second arm: the write committed on its first attempt, the SDK's
own reconnect path
+ // replayed it, and the server's client-table dedup caught the replay
- not a fault.
+ // `to_kafka_error_code`'s shared mapping deliberately doesn't
special-case this - it's a
+ // write-only fact, checked here, at the one write this bridge makes,
rather than assumed
+ // true for the reads that share that mapping too.
+ Ok(TopicCreationOutcome::Created)
+ | Err(BridgeError::Iggy(IggyError::RequestAlreadyApplied)) =>
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(error_message_for(&err)))),
+ }
+}
+
+/// Error text for `CreatableTopicResult.error_message`, without re-embedding
the topic name:
+/// `result.name` (`CreatableTopicResult::with_name`) already carries it, so
`err.to_string()`'s
+/// own embedded copy for these two variants would double the per-topic
response cost for a name
+/// the client already sent and already has back. Validation runs before any
bridge I/O, so this
+/// is a purely local, zero-round-trip amplification if left in - 100 topics
named with a maximal
+/// legal length build a response roughly twice the size the name alone would
justify.
+fn error_message_for(err: &BridgeError) -> String {
+ match err {
+ BridgeError::InvalidKafkaTopicName { reason, .. } => reason.clone(),
+ BridgeError::InvalidPartitionCount { .. } => {
+ "partition count must be at least 1".to_string()
+ }
+ other => other.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`.
+///
+/// An assignment's own partition indices are checked too, not just its
length: real Kafka
+/// requires the key set to be exactly `0..assignments.len()`, each index
appearing once, and
+/// rejects anything else - a duplicate or non-consecutive index (`{5: [...],
7: [...]}`) - with
+/// `INVALID_REPLICA_ASSIGNMENT` (39), the one condition that code exists for.
This bridge doesn't
+/// model per-partition replica placement, so only the index set is checked,
not each entry's
+/// replica list.
+///
+/// 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);
+ }
+ if !assignment_indices_are_consecutive_from_zero(&topic.assignments) {
Review Comment:
The indices are checked now, but each assignment's `broker_ids` never is.
Real Kafka's `ReplicationControlManager` answers `INVALID_REPLICA_ASSIGNMENT`
(39) for an empty replica list, a broker named twice, and a broker that isn't
registered. Here `[]`, `[1, 1]` and `[7]` all answer `NONE` and create the
topic, although this gateway only advertises node 1. The test helper at
`create_topics_real_bridge_tests.rs:91` writes `broker_id 0`, so the suite
currently asserts `NONE` for assignments a real broker would reject.
##########
gateways/kafka/src/bridge/iggy_bridge/topics.rs:
##########
@@ -217,4 +255,140 @@ impl IggyBridge {
Err(err) => Err(err),
}
}
+
+ /// Looks up `kafka_topic`, resolved through the configured
+ /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping), without
creating it.
+ ///
+ /// Returns `Ok(None)` when either the mapped stream or the mapped topic
doesn't exist -
+ /// callers (`CreateTopics`' existence check, `Metadata`'s lookup) treat
both the same way:
+ /// nothing answers to this Kafka-side name yet.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails
Kafka's own
+ /// topic-naming rules. Returns [`BridgeError::Timeout`] if a call takes
longer than
+ /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] for connectivity/auth
failures.
+ pub async fn get_kafka_topic(
+ &self,
+ kafka_topic: &str,
+ ) -> Result<Option<TopicDetails>, BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) =
self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id =
Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ let topic_id =
Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ // No separate get_stream probe: get_topic already answers Ok(None)
when the stream
+ // itself is missing (see high_watermarks' own doc on this same fact),
so a probe first
+ // would just pay a second round trip to learn something this one call
already tells us.
+ with_request_timeout(self.client.get_topic(&stream_id,
&topic_id)).await
+ }
+
+ /// Creates the Iggy stream/topic backing `kafka_topic`, or reports that
it already exists.
+ ///
+ /// Atomic from this call's perspective, unlike a separate existence check
+ /// ([`Self::get_kafka_topic`]) followed by
[`Self::ensure_stream_and_topic`]: that sequence
+ /// has a TOCTOU window between the two calls, and
`ensure_stream_and_topic`'s own idempotent
+ /// contract would then absorb a second concurrent caller's create into a
silent `Ok`, so both
+ /// callers see success for a `CreateTopics` request Kafka promises
exactly one `NONE` for.
+ /// Here, the create attempt itself is the existence check: no separate
read precedes it, and
+ /// [`TopicCreationOutcome::AlreadyExists`] comes from the server's own
rejection of the write,
+ /// not from an earlier read that could already be stale by the time this
call's write lands.
+ ///
+ /// # Errors
+ ///
+ /// Same as [`Self::ensure_stream_and_topic`], except an already-existing
topic is reported as
+ /// [`TopicCreationOutcome::AlreadyExists`] rather than
[`BridgeError::PartitionCountMismatch`].
+ /// `CreateTopics` is not an upsert, so a pre-existing topic is never
itself an error here,
+ /// regardless of whether its partition count matches `partition_count`.
+ pub async fn create_kafka_topic(
+ &self,
+ kafka_topic: &str,
+ partition_count: u32,
+ ) -> Result<TopicCreationOutcome, BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ if partition_count == 0 {
+ return Err(BridgeError::InvalidPartitionCount {
+ kafka_topic: kafka_topic.to_string(),
+ });
+ }
+ let (stream_name, topic_name) =
self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id = self.ensure_stream(stream_name).await?;
+
+ let options = TopicCreateOptions {
+ partitions_count: Some(partition_count),
+ ..TopicCreateOptions::default()
+ };
+ match with_request_timeout(self.client.create_topic(&stream_id,
topic_name, &options)).await
+ {
+ Ok(created) => {
+ info!("created Iggy topic '{topic_name}' with
{partition_count} partitions");
+ // Same postcondition check ensure_topic's own create branch
makes: partitions_count
+ // is a hard argument (Some(partition_count), never None), so
a mismatch here means
+ // a future server-side clamp/cap, not a client input problem
- fails loudly instead
+ // of silently reporting Created under a broken contract.
+ if created.partitions_count != partition_count {
+ return Err(BridgeError::PartitionCountMismatch {
+ topic: kafka_topic.to_string(),
+ existing: created.partitions_count,
+ requested: partition_count,
+ });
+ }
+ Ok(TopicCreationOutcome::Created)
+ }
+ Err(BridgeError::Iggy(IggyError::TopicNameAlreadyExists(_, _))) =>
{
+ Ok(TopicCreationOutcome::AlreadyExists)
+ }
+ Err(err) => Err(err),
+ }
+ }
+
+ /// Every Kafka-visible topic: the target of every configured
+ /// [`TopicMapping`](crate::bridge::topic_map::TopicMapping) override that
actually exists in
+ /// Iggy, plus every topic in the default stream that isn't itself one of
those override
+ /// targets - checked so an overridden topic is never listed twice, once
under its Kafka-side
+ /// name and once under its raw Iggy name.
+ ///
+ /// An Iggy stream this bridge has no mapping rule pointing at (neither
the default stream nor
+ /// any override's target) holds data no Kafka client ever named -
deliberately excluded, the
+ /// same way a real Kafka broker never reports storage it doesn't own.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::Timeout`] if a call takes longer than
`REQUEST_TIMEOUT`. Returns
+ /// [`BridgeError::Iggy`] for connectivity/auth failures.
+ pub async fn list_kafka_topics(&self) -> Result<Vec<KafkaTopicMetadata>,
BridgeError> {
+ let default_stream = self.config.topic_mapping.default_stream();
+ let mut default_stream_override_targets: HashSet<&str> =
HashSet::new();
+ let mut results = Vec::new();
+
+ for (kafka_topic, over) in self.config.topic_mapping.overrides() {
+ if over.stream == default_stream {
+ default_stream_override_targets.insert(over.topic.as_str());
+ }
+ if let Some(details) = self.get_kafka_topic(kafka_topic).await? {
+ results.push(KafkaTopicMetadata {
+ kafka_topic: kafka_topic.to_string(),
+ partitions_count: details.partitions_count,
+ });
+ }
+ }
+
+ let default_stream_id =
Identifier::named(default_stream).map_err(BridgeError::Iggy)?;
+ if with_request_timeout(self.client.get_stream(&default_stream_id))
+ .await?
+ .is_some()
+ {
+ let topics =
with_request_timeout(self.client.get_topics(&default_stream_id)).await?;
+ for topic in topics {
+ if
default_stream_override_targets.contains(topic.name.as_str()) {
Review Comment:
This skips default-stream topics that are an override's *target*, but not
ones whose name is an override *key*. With overrides `foo → (kafka, bar)` and
`bar → (other, x)` (both accepted by `TopicMapping::new`) and Iggy topics
`kafka/foo` and `kafka/bar`, the override loop emits `foo` backed by
`kafka/bar`, and this loop emits `kafka/foo` as a second `foo` with a different
partition count. Real Kafka never repeats a name, and `kafka/foo` can't be
reached by name anyway since `foo` routes to `kafka/bar`. Also skipping names
that are override keys would fix it.
--
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]
