hubcio commented on code in PR #3886: URL: https://github.com/apache/iggy/pull/3886#discussion_r4070717289
########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1072 @@ +// 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::{parse_duration, retry_backoff}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + fmt, + 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 MAX_BATCH_SIZE: u32 = 10_000; +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>, + #[serde(default)] + pub malformed_message_policy: MalformedMessagePolicy, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MalformedMessagePolicy { + #[default] + Block, + DropHeaders, +} + +impl fmt::Display for MalformedMessagePolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Block => formatter.write_str("block"), + Self::DropHeaders => formatter.write_str("drop_headers"), + } + } +} + +/// 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, Default)] +struct PollCycleCounts { + successful_upstream_polls: u64, + upstream_poll_errors: u64, + conversion_errors: u64, +} + +impl PollCycleCounts { + fn total_errors(&self) -> u64 { + self.upstream_poll_errors + self.conversion_errors + } + + fn requires_backoff(&self) -> bool { + self.upstream_poll_errors > 0 && self.successful_upstream_polls == 0 + } +} + +#[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_failed_poll_cycles: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, + 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); + let malformed_message_policy = config.malformed_message_policy; + + 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_failed_poll_cycles: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + malformed_message_policy, + 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> { + validate_batch_size(self.batch_size)?; + + 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: {}, malformed message policy: {}", + self.id, + self.partitions.len(), + self.initial_offset, + self.poll_interval, + self.batch_size, + self.malformed_message_policy + ); + 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 failed_cycles = self.consecutive_failed_poll_cycles.load(Ordering::Relaxed); + if failed_cycles > 0 { + let delay = retry_backoff( + self.retry_interval, + failed_cycles as u32, + self.max_retry_interval, + ); + debug!( + "Backing off for {delay:?} after {failed_cycles} consecutive poll cycles with no \ + successful partitions for \ + {CONNECTOR_NAME} connector ID: {}", + self.id + ); + sleep(delay).await; + } + + let consumer = Consumer::default(); + + let mut candidate_state = self.state.lock().await.clone(); + + let mut messages = Vec::with_capacity(self.batch_size as usize); + let mut cycle_counts = PollCycleCounts::default(); + + for &partition_id in &self.partitions { + let strategy = next_strategy( + self.initial_offset, + candidate_state.offsets.get(&partition_id).copied(), + ); + let polled = client + .poll_messages( + stream_id, + topic_id, + Some(partition_id), + &consumer, + &strategy, + self.batch_size, + false, + ) + .await; + + match polled { + Ok(polled) => { + cycle_counts.successful_upstream_polls += 1; + if polled.messages.is_empty() { + continue; + } + let PartitionBuildOutcome { + messages: partition_messages, + last_produced_offset, + errors: conversion_errors, + } = build_produced_messages( + &polled.messages, + self.include_user_headers, + self.malformed_message_policy, + ); + + cycle_counts.conversion_errors += conversion_errors.len() as u64; + for conversion_error in conversion_errors { + error!( + "Failed to convert upstream message at offset {} on partition \ + {partition_id} for {CONNECTOR_NAME} connector ID: {}. \ + Malformed message policy: {}. {}", + conversion_error.offset, + self.id, + self.malformed_message_policy, + conversion_error.error + ); + } + + messages.extend(partition_messages); + if let Some(last_produced_offset) = last_produced_offset { + candidate_state + .offsets + .insert(partition_id, last_produced_offset); + } + } + Err(IggyError::InvalidOffset(offset)) => { + cycle_counts.upstream_poll_errors += 1; + warn!( + "Saved offset {offset} for partition {partition_id} no longer valid \ + for {CONNECTOR_NAME} connector ID: {}, resetting to initial offset", + self.id + ); + candidate_state.offsets.remove(&partition_id); + } + Err(poll_error) => { + cycle_counts.upstream_poll_errors += 1; + error!( + "Failed to poll partition {partition_id} for {CONNECTOR_NAME} \ + connector ID: {}: {poll_error}", + self.id + ); + } + } + } + + candidate_state.messages_synced += messages.len() as u64; + candidate_state.errors_count += cycle_counts.total_errors(); + let total_synced = candidate_state.messages_synced; + let persisted_state = self.serialize_state(&candidate_state).ok_or_else(|| { + Error::Serialization("failed to serialize Iggy source state".to_string()) + })?; + *self.pending_state.lock().await = Some(candidate_state); + + if cycle_counts.requires_backoff() { + self.consecutive_failed_poll_cycles + .fetch_add(1, Ordering::Relaxed); + } else { + self.consecutive_failed_poll_cycles + .store(0, Ordering::Relaxed); + } + + if self.verbose { + info!( + "{CONNECTOR_NAME} connector ID: {} polled {} messages from {} partition(s). \ + Total synced: {}, poll errors in cycle: {}, conversion errors in cycle: {}", + self.id, + messages.len(), + self.partitions.len(), + total_synced, + cycle_counts.upstream_poll_errors, + cycle_counts.conversion_errors + ); + } else { + debug!( + "{CONNECTOR_NAME} connector ID: {} polled {} messages from {} partition(s). \ + Total synced: {}, poll errors in cycle: {}, conversion errors in cycle: {}", + self.id, + messages.len(), + self.partitions.len(), + total_synced, + cycle_counts.upstream_poll_errors, + cycle_counts.conversion_errors + ); + } + + Ok(ProducedMessages { + schema: Schema::Raw, + messages, + state: Some(persisted_state), + }) + } + + async fn on_batch_result(&self, result: SourceBatchResult) -> Result<(), Error> { + let candidate_state = self.pending_state.lock().await.take(); + if result == SourceBatchResult::Ack + && let Some(candidate_state) = candidate_state + { + *self.state.lock().await = candidate_state; + } + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + if let Some(client) = self.client.take() + && let Err(e) = client.disconnect().await + { + warn!( + "Failed to disconnect from upstream cluster for {CONNECTOR_NAME} \ + connector ID: {}: {e}", + self.id + ); + } + + let state = self.state.lock().await; + info!( + "{CONNECTOR_NAME} connector ID: {} closed. Total messages synced: {}, total errors: {}", + self.id, state.messages_synced, state.errors_count + ); + Ok(()) + } +} + +fn validate_batch_size(batch_size: u32) -> Result<(), Error> { + if !(1..=MAX_BATCH_SIZE).contains(&batch_size) { + return Err(Error::InvalidConfigValue(format!( + "batch_size must be between 1 and {MAX_BATCH_SIZE}, got {batch_size}" + ))); + } + Ok(()) +} + +fn next_strategy(initial: InitialOffset, saved_offset: Option<u64>) -> PollingStrategy { + match saved_offset { + Some(offset) => PollingStrategy::offset(offset.saturating_add(1)), + None => match initial { + InitialOffset::Earliest => PollingStrategy::first(), + InitialOffset::Latest => PollingStrategy::last(), + InitialOffset::Offset(offset) => PollingStrategy::offset(offset), + }, + } +} + +#[derive(Debug)] +struct PartitionBuildOutcome { + messages: Vec<ProducedMessage>, + last_produced_offset: Option<u64>, + errors: Vec<MessageBuildError>, +} + +#[derive(Debug)] +struct MessageBuildError { + offset: u64, + error: Error, +} + +fn build_produced_messages( + messages: &[IggyMessage], + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, +) -> PartitionBuildOutcome { + let mut produced_messages = Vec::with_capacity(messages.len()); + let mut last_produced_offset = None; + let mut errors = Vec::new(); + + for message in messages { + let headers = if include_user_headers { + match message.user_headers_map() { Review Comment: warning: header bytes that fail to parse make `user_headers_map` return `Ok(None)`, so the message ships without headers and the offset advances. a present `user_headers` with a `None` map is that case, so treat it as a build error and apply `malformed_message_policy`. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1072 @@ +// 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::{parse_duration, retry_backoff}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + fmt, + 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 MAX_BATCH_SIZE: u32 = 10_000; +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>, + #[serde(default)] + pub malformed_message_policy: MalformedMessagePolicy, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MalformedMessagePolicy { + #[default] + Block, + DropHeaders, +} + +impl fmt::Display for MalformedMessagePolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Block => formatter.write_str("block"), + Self::DropHeaders => formatter.write_str("drop_headers"), + } + } +} + +/// 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, Default)] +struct PollCycleCounts { + successful_upstream_polls: u64, + upstream_poll_errors: u64, + conversion_errors: u64, +} + +impl PollCycleCounts { + fn total_errors(&self) -> u64 { + self.upstream_poll_errors + self.conversion_errors + } + + fn requires_backoff(&self) -> bool { + self.upstream_poll_errors > 0 && self.successful_upstream_polls == 0 + } +} + +#[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_failed_poll_cycles: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, + 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); + let malformed_message_policy = config.malformed_message_policy; + + 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_failed_poll_cycles: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + malformed_message_policy, + 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> { + validate_batch_size(self.batch_size)?; + + 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: {}, malformed message policy: {}", + self.id, + self.partitions.len(), + self.initial_offset, + self.poll_interval, + self.batch_size, + self.malformed_message_policy + ); + 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 failed_cycles = self.consecutive_failed_poll_cycles.load(Ordering::Relaxed); + if failed_cycles > 0 { + let delay = retry_backoff( + self.retry_interval, + failed_cycles as u32, + self.max_retry_interval, + ); + debug!( + "Backing off for {delay:?} after {failed_cycles} consecutive poll cycles with no \ + successful partitions for \ + {CONNECTOR_NAME} connector ID: {}", + self.id + ); + sleep(delay).await; + } + + let consumer = Consumer::default(); + + let mut candidate_state = self.state.lock().await.clone(); + + let mut messages = Vec::with_capacity(self.batch_size as usize); + let mut cycle_counts = PollCycleCounts::default(); + + for &partition_id in &self.partitions { + let strategy = next_strategy( + self.initial_offset, + candidate_state.offsets.get(&partition_id).copied(), + ); + let polled = client + .poll_messages( + stream_id, + topic_id, + Some(partition_id), + &consumer, + &strategy, + self.batch_size, + false, + ) + .await; + + match polled { + Ok(polled) => { + cycle_counts.successful_upstream_polls += 1; + if polled.messages.is_empty() { + continue; + } + let PartitionBuildOutcome { + messages: partition_messages, + last_produced_offset, + errors: conversion_errors, + } = build_produced_messages( + &polled.messages, + self.include_user_headers, + self.malformed_message_policy, + ); + + cycle_counts.conversion_errors += conversion_errors.len() as u64; + for conversion_error in conversion_errors { + error!( + "Failed to convert upstream message at offset {} on partition \ + {partition_id} for {CONNECTOR_NAME} connector ID: {}. \ + Malformed message policy: {}. {}", + conversion_error.offset, + self.id, + self.malformed_message_policy, + conversion_error.error + ); + } + + messages.extend(partition_messages); + if let Some(last_produced_offset) = last_produced_offset { + candidate_state + .offsets + .insert(partition_id, last_produced_offset); + } + } + Err(IggyError::InvalidOffset(offset)) => { Review Comment: warning: this arm can never fire - the server only raises `InvalidOffset` from the consumer-offset store path, and the connector keeps its offsets in its own state file. delete the arm and the README:98 reset claim, or make the reset real. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1072 @@ +// 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::{parse_duration, retry_backoff}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + fmt, + 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 MAX_BATCH_SIZE: u32 = 10_000; +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>, + #[serde(default)] + pub malformed_message_policy: MalformedMessagePolicy, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MalformedMessagePolicy { + #[default] + Block, + DropHeaders, +} + +impl fmt::Display for MalformedMessagePolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Block => formatter.write_str("block"), + Self::DropHeaders => formatter.write_str("drop_headers"), + } + } +} + +/// 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, Default)] +struct PollCycleCounts { + successful_upstream_polls: u64, + upstream_poll_errors: u64, + conversion_errors: u64, +} + +impl PollCycleCounts { + fn total_errors(&self) -> u64 { + self.upstream_poll_errors + self.conversion_errors + } + + fn requires_backoff(&self) -> bool { + self.upstream_poll_errors > 0 && self.successful_upstream_polls == 0 + } +} + +#[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_failed_poll_cycles: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, + 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); + let malformed_message_policy = config.malformed_message_policy; + + 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_failed_poll_cycles: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + malformed_message_policy, + 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) => { Review Comment: warning: if the name is already taken (code 1012 for stream, 2013 for topic) `open()` fails and nothing retries, so a lost race leaves the connector down until an operator restarts it. treat already-exists from create as success. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1072 @@ +// 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::{parse_duration, retry_backoff}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + fmt, + 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 MAX_BATCH_SIZE: u32 = 10_000; +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>, + #[serde(default)] + pub malformed_message_policy: MalformedMessagePolicy, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MalformedMessagePolicy { + #[default] + Block, + DropHeaders, +} + +impl fmt::Display for MalformedMessagePolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Block => formatter.write_str("block"), + Self::DropHeaders => formatter.write_str("drop_headers"), + } + } +} + +/// 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, Default)] +struct PollCycleCounts { + successful_upstream_polls: u64, + upstream_poll_errors: u64, + conversion_errors: u64, +} + +impl PollCycleCounts { + fn total_errors(&self) -> u64 { + self.upstream_poll_errors + self.conversion_errors + } + + fn requires_backoff(&self) -> bool { + self.upstream_poll_errors > 0 && self.successful_upstream_polls == 0 + } +} + +#[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_failed_poll_cycles: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, + 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); + let malformed_message_policy = config.malformed_message_policy; + + 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_failed_poll_cycles: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + malformed_message_policy, + 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> { + validate_batch_size(self.batch_size)?; + + 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 Review Comment: critical: if the upstream topic is re-created, saved offsets point past the new frontier, so every poll returns an empty batch with no error and that history is never copied. store the topic `created_at` and clear the offsets on change. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1072 @@ +// 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::{parse_duration, retry_backoff}, + source::SourceBatchResult, + source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::{ + collections::HashMap, + fmt, + 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 MAX_BATCH_SIZE: u32 = 10_000; +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>, + #[serde(default)] + pub malformed_message_policy: MalformedMessagePolicy, + pub retry_interval: Option<String>, + pub max_retry_interval: Option<String>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MalformedMessagePolicy { + #[default] + Block, + DropHeaders, +} + +impl fmt::Display for MalformedMessagePolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Block => formatter.write_str("block"), + Self::DropHeaders => formatter.write_str("drop_headers"), + } + } +} + +/// 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, Default)] +struct PollCycleCounts { + successful_upstream_polls: u64, + upstream_poll_errors: u64, + conversion_errors: u64, +} + +impl PollCycleCounts { + fn total_errors(&self) -> u64 { + self.upstream_poll_errors + self.conversion_errors + } + + fn requires_backoff(&self) -> bool { + self.upstream_poll_errors > 0 && self.successful_upstream_polls == 0 + } +} + +#[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_failed_poll_cycles: AtomicU64, + initial_offset: InitialOffset, + batch_size: u32, + include_user_headers: bool, + malformed_message_policy: MalformedMessagePolicy, + 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); + let malformed_message_policy = config.malformed_message_policy; + + 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_failed_poll_cycles: AtomicU64::new(0), + initial_offset, + batch_size, + include_user_headers, + malformed_message_policy, + 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> { + validate_batch_size(self.batch_size)?; + + 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: {}, malformed message policy: {}", + self.id, + self.partitions.len(), + self.initial_offset, + self.poll_interval, + self.batch_size, + self.malformed_message_policy + ); + 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 failed_cycles = self.consecutive_failed_poll_cycles.load(Ordering::Relaxed); + if failed_cycles > 0 { + let delay = retry_backoff( + self.retry_interval, + failed_cycles as u32, + self.max_retry_interval, + ); + debug!( + "Backing off for {delay:?} after {failed_cycles} consecutive poll cycles with no \ + successful partitions for \ + {CONNECTOR_NAME} connector ID: {}", + self.id + ); + sleep(delay).await; + } + + let consumer = Consumer::default(); + + let mut candidate_state = self.state.lock().await.clone(); + + let mut messages = Vec::with_capacity(self.batch_size as usize); + let mut cycle_counts = PollCycleCounts::default(); + + for &partition_id in &self.partitions { + let strategy = next_strategy( + self.initial_offset, + candidate_state.offsets.get(&partition_id).copied(), + ); + let polled = client + .poll_messages( + stream_id, + topic_id, + Some(partition_id), + &consumer, + &strategy, + self.batch_size, + false, + ) + .await; + + match polled { + Ok(polled) => { + cycle_counts.successful_upstream_polls += 1; + if polled.messages.is_empty() { + continue; + } + let PartitionBuildOutcome { + messages: partition_messages, + last_produced_offset, + errors: conversion_errors, + } = build_produced_messages( + &polled.messages, + self.include_user_headers, + self.malformed_message_policy, + ); + + cycle_counts.conversion_errors += conversion_errors.len() as u64; + for conversion_error in conversion_errors { + error!( + "Failed to convert upstream message at offset {} on partition \ + {partition_id} for {CONNECTOR_NAME} connector ID: {}. \ + Malformed message policy: {}. {}", + conversion_error.offset, + self.id, + self.malformed_message_policy, + conversion_error.error + ); + } + + messages.extend(partition_messages); + if let Some(last_produced_offset) = last_produced_offset { + candidate_state + .offsets + .insert(partition_id, last_produced_offset); + } + } + Err(IggyError::InvalidOffset(offset)) => { + cycle_counts.upstream_poll_errors += 1; + warn!( + "Saved offset {offset} for partition {partition_id} no longer valid \ + for {CONNECTOR_NAME} connector ID: {}, resetting to initial offset", + self.id + ); + candidate_state.offsets.remove(&partition_id); + } + Err(poll_error) => { + cycle_counts.upstream_poll_errors += 1; + error!( + "Failed to poll partition {partition_id} for {CONNECTOR_NAME} \ + connector ID: {}: {poll_error}", + self.id + ); + } + } + } + + candidate_state.messages_synced += messages.len() as u64; + candidate_state.errors_count += cycle_counts.total_errors(); + let total_synced = candidate_state.messages_synced; + let persisted_state = self.serialize_state(&candidate_state).ok_or_else(|| { + Error::Serialization("failed to serialize Iggy source state".to_string()) + })?; + *self.pending_state.lock().await = Some(candidate_state); + + if cycle_counts.requires_backoff() { + self.consecutive_failed_poll_cycles + .fetch_add(1, Ordering::Relaxed); + } else { + self.consecutive_failed_poll_cycles + .store(0, Ordering::Relaxed); + } + + if self.verbose { + info!( + "{CONNECTOR_NAME} connector ID: {} polled {} messages from {} partition(s). \ + Total synced: {}, poll errors in cycle: {}, conversion errors in cycle: {}", + self.id, + messages.len(), + self.partitions.len(), + total_synced, + cycle_counts.upstream_poll_errors, + cycle_counts.conversion_errors + ); + } else { + debug!( + "{CONNECTOR_NAME} connector ID: {} polled {} messages from {} partition(s). \ + Total synced: {}, poll errors in cycle: {}, conversion errors in cycle: {}", + self.id, + messages.len(), + self.partitions.len(), + total_synced, + cycle_counts.upstream_poll_errors, + cycle_counts.conversion_errors + ); + } + + Ok(ProducedMessages { Review Comment: warning: the poll returns state on every cycle, so an idle connector writes and syncs the same bytes on each poll interval. return `state: None` when nothing changed, like `postgres_source` does. ########## core/connectors/sources/iggy_source/README.md: ########## @@ -0,0 +1,130 @@ +# Iggy Source Connector + +The Iggy source connector replicates a topic from an upstream Apache Iggy +cluster into a stream and topic managed by the connectors runtime. It polls +every upstream partition and persists the last acknowledged offset for each +partition so replication can resume after a restart. + +## Features + +- Polls every partition of one upstream topic. +- Preserves payload bytes and non-zero message IDs. +- Preserves user headers by default, with an option to omit them. +- Stores per-partition offsets in the connector runtime's state storage. +- Retries fully failed polling cycles with exponential backoff and jitter. +- Creates the configured upstream stream or topic when it does not exist. + +## Configuration + +```toml +type = "source" +key = "iggy" +enabled = true +version = 0 +name = "Iggy source" +path = "target/release/libiggy_connector_iggy_source" +verbose = false +benchmark = false + +[[streams]] +stream = "downstream_stream" +topic = "downstream_topic" +schema = "raw" +batch_length = 100 +linger_time = "5ms" + +[plugin_config] +connection_string = "iggy+tcp://iggy:[email protected]:8090" +upstream_stream = "upstream_stream" +upstream_topic = "upstream_topic" +poll_interval = "1s" +batch_size = 100 +initial_offset = "earliest" +include_user_headers = true +malformed_message_policy = "block" +retry_interval = "1s" +max_retry_interval = "60s" +verbose_logging = false +``` + +Use `schema = "raw"` on the downstream stream to replicate payload bytes +without decoding or re-encoding them. The runtime-level `batch_length` controls +downstream producer batching, while `plugin_config.batch_size` controls how many +messages are requested from each upstream partition in one poll cycle. + +### Plugin Fields + +| Field | Required | Default | Description | +| --- | --- | --- | --- | +| `connection_string` | yes | none | Connection string for the upstream Iggy cluster. | +| `upstream_stream` | yes | none | Name of the stream to replicate from. | +| `upstream_topic` | yes | none | Name of the topic to replicate from. | +| `poll_interval` | no | `2s` | Delay before each upstream poll cycle. | +| `batch_size` | no | `100` | Maximum messages requested from each upstream partition per poll cycle. Must be between `1` and `10,000`. | +| `initial_offset` | no | `earliest` | Starting position for a partition without a saved offset. Accepts `earliest`, `latest`, or an absolute numeric offset. | +| `include_user_headers` | no | `true` | Copy user headers to downstream messages. | +| `malformed_message_policy` | no | `block` | Handling for messages with unparsable user headers: `block` retries from the failed offset; `drop_headers` forwards the payload without those headers. | +| `retry_interval` | no | `1s` | Base delay used for exponential backoff when every partition poll in a cycle fails. | +| `max_retry_interval` | no | `60s` | Maximum delay between poll retries. | +| `verbose_logging` | no | `false` | Log per-cycle connector details at info level instead of debug level. | + +The `connection_string` can use any transport supported by the Rust client, +including TCP, QUIC, HTTP, and WebSocket. It may contain credentials, so keep +the connector configuration private or supply the value through the +`IGGY_CONNECTORS_SOURCE_IGGY_PLUGIN_CONFIG_CONNECTION_STRING` environment +variable. + +## Offset and Delivery Semantics + +For each partition, the connector resumes at one offset after the last saved +offset. `initial_offset` is used only when that partition has no saved offset: + +- `earliest` starts with the oldest available message. +- `latest` requests up to `batch_size` of the most recent messages. +- A numeric value starts at that absolute offset. + +The connector stages new offsets while polling. It commits them only after the +runtime sends the complete downstream batch and saves the connector state. If +the batch is rejected, the previous offsets remain committed and the messages +are eligible for replay. Consumers should therefore tolerate duplicate +delivery after failures or crashes. Review Comment: warning: this replay only happens while the source is running - five consecutive send failures stop it and the runtime never respawns it, so an outage longer than that needs a manual restart. -- 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]
