ryerraguntla commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4093457921
##########
gateways/kafka/src/bridge/iggy_bridge/topics.rs:
##########
@@ -217,4 +255,151 @@ 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 and isn't itself named the same as an override key - checked
so an overridden
+ /// topic is never listed twice, once under its Kafka-side name and once
under its raw Iggy
+ /// name, and so a raw default-stream topic never masquerades under a
Kafka-side name an
+ /// override has already claimed for different data. The second check
matters for a chained
+ /// override (`foo -> (kafka, bar)`, `bar -> (other, x)`): without it, a
raw Iggy topic
+ /// literally named `foo` sitting in the default stream would be reported
a second time under
+ /// the same `foo` name the override loop already emitted (backed by
`kafka/bar`'s data), and
+ /// that second `foo` would be unreachable by name anyway, since
`get_kafka_topic("foo")`
+ /// always resolves through the override to `kafka/bar`, never to the raw
`kafka/foo`.
+ ///
+ /// 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 override_keys: HashSet<&str> = HashSet::new();
+ let mut results = Vec::new();
+
+ for (kafka_topic, over) in self.config.topic_mapping.overrides() {
+ override_keys.insert(kafka_topic);
+ 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))
Review Comment:
dropped the get_stream probe, call get_topics alone — confirmed against
core/server/src/responses.rs::build_get_topics_response's own comment ("Legacy
parity: a missing stream lists
as empty, not StreamNotFound") that this is safe.
--
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]
