ryerraguntla commented on code in PR #4258:
URL: https://github.com/apache/iggy/pull/4258#discussion_r4088476645
##########
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:
Now also collects override keys and skips default-stream topics matching
either set.
--
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]
