github-actions[bot] commented on code in PR #3886: URL: https://github.com/apache/iggy/pull/3886#discussion_r4126670849
########## core/integration/tests/connectors/mod.rs: ########## @@ -24,6 +24,7 @@ mod fixtures; mod http; mod http_config_provider; mod iceberg; +mod iggy_source; Review Comment: warning: `.github/actions/rust/pre-merge/action.yml` maps `iggy_connector_iggy_source` to the suite module `connectors::iggy`, so the new `connectors/iggy_source` suite is skipped with a warning when only the plugin changes. Rename the suite directory and module to `iggy`, or extend that mapping. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1197 @@ +// 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, PartialEq, Eq, Serialize, Deserialize)] +struct State { + offsets: HashMap<u32, u64>, + messages_synced: u64, + errors_count: u64, + topic_created_at: Option<u64>, +} + +impl State { + fn reset_offsets_if_topic_changed(&mut self, created_at: u64) -> bool { + if self.topic_created_at == Some(created_at) { + return false; + } + + // Old offsets can be past the new topic's frontier, where polls return + // empty batches and silently skip its history. + let reset_offsets = !self.offsets.is_empty(); + self.offsets.clear(); + self.topic_created_at = Some(created_at); + reset_offsets + } +} + +#[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, + topic_created_at: None, + })), + 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 + ); + match client.create_stream(&self.config.upstream_stream).await { + Ok(_) | Err(IggyError::StreamNameAlreadyExists(_)) => {} + Err(e) => { + return Err(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 + ); + match client + .create_topic( + stream_id, + &self.config.upstream_topic, + &TopicCreateOptions { + partitions_count: Some(AUTO_CREATED_PARTITIONS_COUNT), + compression_algorithm: Some(CompressionAlgorithm::None), + ..TopicCreateOptions::default() Review Comment: warning: `ensure_stream_and_topic()` passes `TopicCreateOptions::default()`, which sends `durability: Replicated` and leaves the flush threshold at the resolved default of 1024 messages. A topic on the weaker policy can lose an acknowledged write before this connector polls it. Create the upstream topic with `Durability::Persisted` and `messages_required_to_save: Some(1)`, as the runtime does for connector topics. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1197 @@ +// 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, PartialEq, Eq, Serialize, Deserialize)] +struct State { + offsets: HashMap<u32, u64>, + messages_synced: u64, + errors_count: u64, + topic_created_at: Option<u64>, +} + +impl State { + fn reset_offsets_if_topic_changed(&mut self, created_at: u64) -> bool { + if self.topic_created_at == Some(created_at) { + return false; + } + + // Old offsets can be past the new topic's frontier, where polls return + // empty batches and silently skip its history. + let reset_offsets = !self.offsets.is_empty(); + self.offsets.clear(); + self.topic_created_at = Some(created_at); + reset_offsets + } +} + +#[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, + topic_created_at: None, + })), + 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 + ); + match client.create_stream(&self.config.upstream_stream).await { + Ok(_) | Err(IggyError::StreamNameAlreadyExists(_)) => {} + Err(e) => { + return Err(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 + ); + match 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 + { + Ok(_) | Err(IggyError::TopicNameAlreadyExists(_, _)) => {} + Err(e) => { + return Err(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 mut candidate_state = self.state.lock().await.clone(); + let topic = client + .get_topic(stream_id, topic_id) + .await + .map_err(|error| { + Error::Connection(format!("Failed to fetch upstream topic details: {error}")) + })?; + let current_partitions = topic.as_ref().map(|topic| { + if candidate_state.reset_offsets_if_topic_changed(topic.created_at.as_micros()) { + warn!( + "Upstream topic '{}/{}' changed for {CONNECTOR_NAME} connector ID: {}, \ + resetting saved offsets to initial offset", + self.config.upstream_stream, self.config.upstream_topic, self.id + ); + } + topic + .partitions + .iter() + .map(|partition| partition.id) + .collect::<Vec<_>>() + }); + let partitions = current_partitions.as_deref().unwrap_or(&self.partitions); + let consumer = Consumer::default(); + + let mut messages = Vec::with_capacity(self.batch_size as usize); + let mut cycle_counts = PollCycleCounts::default(); + + for &partition_id in 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(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 state_changed = *self.state.lock().await != candidate_state; + let persisted_state = if state_changed { + 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); + Some(persisted_state) + } else { + None + }; + + 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(), + 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(), + partitions.len(), + total_synced, + cycle_counts.upstream_poll_errors, + cycle_counts.conversion_errors + ); + } + + Ok(ProducedMessages { + schema: Schema::Raw, + messages, + state: 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() { + Ok(None) if message.user_headers.is_some() => { + errors.push(MessageBuildError { + offset: message.header.offset, + error: Error::InvalidRecordValue( + "Failed to parse upstream user headers".to_string(), + ), + }); + if malformed_message_policy == MalformedMessagePolicy::Block { + break; + } + None + } + Ok(headers) => headers, + Err(error) => { + errors.push(MessageBuildError { + offset: message.header.offset, + error: Error::InvalidRecordValue(format!( + "Failed to parse upstream user headers: {error}" + )), + }); + if malformed_message_policy == MalformedMessagePolicy::Block { + break; + } + None + } + } + } else { + None + }; + last_produced_offset = Some(message.header.offset); + produced_messages.push(ProducedMessage { + id: (message.header.id != 0).then_some(message.header.id), + headers, + checksum: None, + timestamp: None, + origin_timestamp: (message.header.origin_timestamp != 0) + .then_some(message.header.origin_timestamp), + payload: message.payload.to_vec(), + }); + } + + PartitionBuildOutcome { + messages: produced_messages, + last_produced_offset, + errors, + } +} + +fn redact_connection_string(connection_string: &str) -> String { + let Some(scheme_end) = connection_string.find("://") else { + return "***".to_string(); + }; + let Some(rest) = connection_string.get(scheme_end + 3..) else { + return "***".to_string(); + }; + let Some(at_offset) = rest.find('@') else { Review Comment: nit: `redact_connection_string()` splits the authority at the first `@`, so a password containing `@` leaves its tail in the start-up log line. The connector logs that line before the client rejects the malformed string, so the fragment still reaches the log file. Split at the last `@`, or log the redacted value only after a successful parse. ########## core/connectors/sources/iggy_source/Cargo.toml: ########## @@ -0,0 +1,51 @@ +# 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. + +[package] +name = "iggy_connector_iggy_source" +version = "0.5.0-edge.1" +description = "Iggy source connector replicating a topic from an upstream Apache Iggy cluster" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "replication"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org"; +documentation = "https://iggy.apache.org/docs"; +repository = "https://github.com/apache/iggy"; +readme = "../../README.md" +publish = false + +[package.metadata.cargo-machete] +ignored = ["dashmap"] + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +dashmap = { workspace = true } +iggy = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +rmp-serde = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } Review Comment: nit: Every other connector crate sets `[lints] workspace = true`, and this manifest omits it, so the workspace `warnings = "deny"` does not apply here. Add the same section. ########## core/connectors/sources/iggy_source/README.md: ########## @@ -0,0 +1,134 @@ +# 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, including after the upstream topic is recreated with the same name. 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. + +If the upstream topic is recreated with the same name, the connector clears +its saved offsets and uses `initial_offset` for the new topic. + +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. + +After five consecutive downstream send or state-save failures, the source +stops. An operator must restart it before it can replay the unacknowledged +messages. + +> When `initial_offset = "latest"` and no offset has been committed yet, each +> retry recalculates the current tail of the upstream partition. If a first +> batch is rejected while new messages arrive, messages from that failed batch +> can be skipped. Use `earliest` or a numeric offset when replay continuity is +> required. + +## Message Mapping + +| Upstream field | Downstream behavior | +| --- | --- | +| Payload | Copied without modification when the downstream schema is `raw`. | +| Message ID | Copied when the upstream ID is non-zero. | +| User headers | Copied when `include_user_headers = true`. | +| Offset and partition | Reassigned by the downstream Iggy topic. | +| Timestamp and checksum | Generated for the downstream message. | + +## Operational Notes + +- The connector discovers upstream partitions when it opens. Restart it after Review Comment: nit: The README tells operators to restart the connector after upstream partitions are added, but `poll()` reads the partition list from the topic on every cycle. Replace the restart advice with a note that new partitions are discovered automatically. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1197 @@ +// 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, PartialEq, Eq, Serialize, Deserialize)] +struct State { + offsets: HashMap<u32, u64>, + messages_synced: u64, + errors_count: u64, + topic_created_at: Option<u64>, +} + +impl State { + fn reset_offsets_if_topic_changed(&mut self, created_at: u64) -> bool { + if self.topic_created_at == Some(created_at) { + return false; + } + + // Old offsets can be past the new topic's frontier, where polls return + // empty batches and silently skip its history. + let reset_offsets = !self.offsets.is_empty(); + self.offsets.clear(); + self.topic_created_at = Some(created_at); + reset_offsets + } +} + +#[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, + topic_created_at: None, + })), + 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 + ); + match client.create_stream(&self.config.upstream_stream).await { + Ok(_) | Err(IggyError::StreamNameAlreadyExists(_)) => {} + Err(e) => { + return Err(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 + ); + match 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 + { + Ok(_) | Err(IggyError::TopicNameAlreadyExists(_, _)) => {} + Err(e) => { + return Err(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 mut candidate_state = self.state.lock().await.clone(); + let topic = client + .get_topic(stream_id, topic_id) + .await + .map_err(|error| { + Error::Connection(format!("Failed to fetch upstream topic details: {error}")) + })?; + let current_partitions = topic.as_ref().map(|topic| { + if candidate_state.reset_offsets_if_topic_changed(topic.created_at.as_micros()) { + warn!( + "Upstream topic '{}/{}' changed for {CONNECTOR_NAME} connector ID: {}, \ + resetting saved offsets to initial offset", + self.config.upstream_stream, self.config.upstream_topic, self.id + ); + } + topic + .partitions + .iter() + .map(|partition| partition.id) + .collect::<Vec<_>>() + }); + let partitions = current_partitions.as_deref().unwrap_or(&self.partitions); + let consumer = Consumer::default(); + + let mut messages = Vec::with_capacity(self.batch_size as usize); + let mut cycle_counts = PollCycleCounts::default(); + + for &partition_id in 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); Review Comment: warning: `poll()` extends one batch with up to `batch_size` messages from every partition, so a wide topic holds `partitions × batch_size` messages at once. Each payload is copied into the batch, and the SDK serializes the whole batch again for the FFI call. Cap the messages in a cycle and stop polling partitions once the cap is reached. ########## core/connectors/sources/iggy_source/src/lib.rs: ########## @@ -0,0 +1,1197 @@ +// 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, PartialEq, Eq, Serialize, Deserialize)] +struct State { + offsets: HashMap<u32, u64>, + messages_synced: u64, + errors_count: u64, + topic_created_at: Option<u64>, +} + +impl State { + fn reset_offsets_if_topic_changed(&mut self, created_at: u64) -> bool { + if self.topic_created_at == Some(created_at) { + return false; + } + + // Old offsets can be past the new topic's frontier, where polls return + // empty batches and silently skip its history. + let reset_offsets = !self.offsets.is_empty(); + self.offsets.clear(); + self.topic_created_at = Some(created_at); + reset_offsets + } +} + +#[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(|| { Review Comment: nit: `IggySource::new()` logs `Invalid initial offset None ... defaulting to earliest` at WARN when `initial_offset` is absent, which is the documented default. Warn only when a supplied value fails to parse. -- 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]
