mattp5657 commented on code in PR #3886: URL: https://github.com/apache/iggy/pull/3886#discussion_r3948690071
########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,806 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use async_trait::async_trait; +use iggy::prelude::{ + Client, CompressionAlgorithm, Consumer, Identifier, IggyClient, IggyError, IggyMessage, + MessageClient, PollingStrategy, StreamClient, TopicClient, TopicCreateOptions, +}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + retry::{exponential_backoff, jitter, parse_duration}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + str::FromStr, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use tokio::{sync::Mutex, time::sleep}; +use tracing::{debug, error, info, warn}; + +source_connector!(IggySource); + +const CONNECTOR_NAME: &str = "Iggy source"; +const DEFAULT_POLL_INTERVAL: &str = "2s"; +const DEFAULT_RETRY_INTERVAL: &str = "1s"; +const DEFAULT_MAX_RETRY_INTERVAL: &str = "60s"; +const DEFAULT_BATCH_SIZE: u32 = 100; +const AUTO_CREATED_PARTITIONS_COUNT: u32 = 1; + +/// Configuration for the Iggy source connector, replicating a topic from an +/// upstream Iggy cluster. `connection_string` points at the upstream cluster. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IggySourceConfig { + #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")] + pub connection_string: SecretString, + pub upstream_stream: String, + pub upstream_topic: String, + pub poll_interval: Option<String>, + pub batch_size: Option<u32>, + pub initial_offset: Option<String>, + pub include_user_headers: Option<bool>, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +/// Starting point for a partition that has no saved offset in the state yet. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InitialOffset { + Earliest, + Latest, + Offset(u64), +} + +impl FromStr for InitialOffset { + type Err = (); + + fn from_str(s: &str) -> Result<Self, Self::Err> { + match s.trim().to_lowercase().as_str() { + "earliest" => Ok(Self::Earliest), + "latest" => Ok(Self::Latest), + other => other.parse::<u64>().map(Self::Offset).map_err(|_| ()), + } + } +} + +/// Committed state. `offsets` maps each upstream partition to the offset of +/// the last message confirmed by the runtime after both the downstream send +/// and checkpoint save succeeded. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct State { + offsets: HashMap<u32, u64>, + messages_synced: u64, + errors_count: u64, +} + +#[derive(Debug)] +pub struct IggySource { + id: u32, + config: IggySourceConfig, + client: Option<IggyClient>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + partitions: Vec<u32>, + stream_id: Option<Identifier>, + topic_id: Option<Identifier>, + poll_interval: Duration, + retry_interval: Duration, + max_retry_interval: Duration, + consecutive_failures: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + verbose: bool, +} + +impl IggySource { + pub fn new(id: u32, config: IggySourceConfig, state: Option<ConnectorState>) -> Self { + let verbose = config.verbose_logging.unwrap_or(false); + let restored_state = state + .and_then(|s| s.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|s| { + info!( + "Restored state for {CONNECTOR_NAME} connector ID: {id}. \ + Offsets: {:?}, messages synced: {}, errors: {}", + s.offsets, s.messages_synced, s.errors_count + ); + }); + + let poll_interval = parse_duration(config.poll_interval.as_deref(), DEFAULT_POLL_INTERVAL); + let retry_interval = + parse_duration(config.retry_interval.as_deref(), DEFAULT_RETRY_INTERVAL); + let max_retry_interval = parse_duration( + config.max_retry_interval.as_deref(), + DEFAULT_MAX_RETRY_INTERVAL, + ); + + let initial_offset = config + .initial_offset + .as_deref() + .map(InitialOffset::from_str) + .and_then(Result::ok) + .unwrap_or_else(|| { + warn!( + "Invalid initial offset {:?} for {CONNECTOR_NAME} connector ID: {id}, \ + defaulting to earliest", + config.initial_offset + ); + InitialOffset::Earliest + }); + + let batch_size = config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE); + let include_user_headers = config.include_user_headers.unwrap_or(true); + + IggySource { + id, + config, + client: None, + state: Mutex::new(restored_state.unwrap_or(State { + offsets: HashMap::new(), + messages_synced: 0, + errors_count: 0, + })), + pending_state: Mutex::new(None), + partitions: Vec::new(), + stream_id: None, + topic_id: None, + poll_interval, + retry_interval, + max_retry_interval, + consecutive_failures: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + verbose, + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + async fn ensure_stream_and_topic( + &self, + client: &IggyClient, + stream_id: &Identifier, + topic_id: &Identifier, + ) -> Result<(), Error> { + match client.get_stream(stream_id).await { + Ok(Some(_)) => {} + Ok(None) => { + warn!( + "Upstream stream '{}' does not exist, creating it for {CONNECTOR_NAME} \ + connector ID: {}", + self.config.upstream_stream, self.id + ); + client + .create_stream(&self.config.upstream_stream) + .await + .map_err(|e| { + Error::InitError(format!( + "Failed to create upstream stream '{}': {e}", + self.config.upstream_stream + )) + })?; + } + Err(e) => { + return Err(Error::InitError(format!( + "Failed to check upstream stream '{}': {e}", + self.config.upstream_stream + ))); + } + } + + match client.get_topic(stream_id, topic_id).await { + Ok(Some(_)) => Ok(()), + Ok(None) => { + warn!( + "Upstream topic '{}' does not exist, creating it with \ + {AUTO_CREATED_PARTITIONS_COUNT} partition(s) for {CONNECTOR_NAME} \ + connector ID: {}", + self.config.upstream_topic, self.id + ); + client + .create_topic( + stream_id, + &self.config.upstream_topic, + &TopicCreateOptions { + partitions_count: Some(AUTO_CREATED_PARTITIONS_COUNT), + compression_algorithm: Some(CompressionAlgorithm::None), + ..TopicCreateOptions::default() + }, + ) + .await + .map_err(|e| { + Error::InitError(format!( + "Failed to create upstream topic '{}': {e}", + self.config.upstream_topic + )) + })?; + Ok(()) + } + Err(e) => Err(Error::InitError(format!( + "Failed to check upstream topic '{}': {e}", + self.config.upstream_topic + ))), + } + } +} + +#[async_trait] +impl Source for IggySource { + async fn open(&mut self) -> Result<(), Error> { + let redacted = redact_connection_string(self.config.connection_string.expose_secret()); + info!( + "Opening {CONNECTOR_NAME} connector ID: {}, upstream: {}/{} at {}", + self.id, self.config.upstream_stream, self.config.upstream_topic, redacted + ); + + let client = + IggyClient::from_connection_string(self.config.connection_string.expose_secret()) + .map_err(|e| { + Error::InitError(format!("Failed to parse upstream connection string: {e}")) + })?; + + client.connect().await.map_err(|e| { + Error::InitError(format!("Failed to connect to upstream Iggy cluster: {e}")) + })?; + + let stream_id = Identifier::named(&self.config.upstream_stream).map_err(|_| { + Error::InvalidConfigValue(format!( + "Invalid upstream stream name '{}'", + self.config.upstream_stream + )) + })?; + let topic_id = Identifier::named(&self.config.upstream_topic).map_err(|_| { + Error::InvalidConfigValue(format!( + "Invalid upstream topic name '{}'", + self.config.upstream_topic + )) + })?; + + self.stream_id = Some(stream_id.clone()); + self.topic_id = Some(topic_id.clone()); + + self.ensure_stream_and_topic(&client, &stream_id, &topic_id) + .await?; + + let topic = client + .get_topic(&stream_id, &topic_id) + .await + .map_err(|e| Error::InitError(format!("Failed to fetch upstream topic details: {e}")))? + .ok_or_else(|| { + Error::InitError(format!( + "Upstream topic '{}/{}' not found after creation", + self.config.upstream_stream, self.config.upstream_topic + )) + })?; + self.partitions = topic + .partitions + .iter() + .map(|partition| partition.id) + .collect(); + + self.client = Some(client); + info!( + "Opened {CONNECTOR_NAME} connector ID: {}, partitions: {}, initial offset: {:?}, \ + poll interval: {:?}, batch size: {}", + self.id, + self.partitions.len(), + self.initial_offset, + self.poll_interval, + self.batch_size + ); + Ok(()) + } + + async fn poll(&self) -> Result<ProducedMessages, Error> { + sleep(self.poll_interval).await; + + let client = self + .client + .as_ref() + .ok_or_else(|| Error::InitError("Upstream client not connected".to_string()))?; + let stream_id = self + .stream_id + .as_ref() + .ok_or_else(|| Error::InitError("Upstream stream not initialized".to_string()))?; + let topic_id = self + .topic_id + .as_ref() + .ok_or_else(|| Error::InitError("Upstream topic not initialized".to_string()))?; + + let failures = self.consecutive_failures.load(Ordering::Relaxed); Review Comment: `consecutive_failures` is incremented *after* a failure and then passed directly as the 0-based `attempt` argument to `exponential_backoff(base, attempt, max)` on the next cycle. The first retry therefore computes `base × 2^1` instead of `base × 2^0`, doubling every step of the backoff schedule versus what `retry_interval` configures. -- 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]
