seokjin0414 commented on code in PR #3799: URL: https://github.com/apache/iggy/pull/3799#discussion_r4114223625
########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1037 @@ +// 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. + +mod mapping; + +use async_trait::async_trait; +use fluss::client::{EARLIEST_OFFSET, FlussConnection, LogScanner}; +use fluss::config::Config; +use fluss::metadata::{DataField, TablePath}; +use fluss::record::ScanRecords; +use fluss::rpc::message::OffsetSpec; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, info, warn}; + +source_connector!(FlussSource); + +const CONNECTOR_NAME: &str = "Apache Fluss source"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(5); +const LOG_TABLE_TYPE: &str = "log"; +const JSON_PAYLOAD_FORMAT: &str = "json"; +const METADATA_BUCKET: &str = "_fluss_bucket"; +const METADATA_OFFSET: &str = "_fluss_offset"; +const METADATA_TIMESTAMP: &str = "_fluss_timestamp"; +const NANOS_PER_MILLI: u64 = 1_000_000; +/// A rewind is local scanner bookkeeping unless the client has to refresh the table's +/// metadata first. An error from `on_batch_result` stops the source, so that refresh gets a +/// few attempts before the failure is reported. +const REWIND_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(1), +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. A primary-key table streams a changelog whose change + /// types need their own mapping onto messages, so the value is validated rather than + /// silently ignored. + pub table_type: Option<String>, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option<String>, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option<Vec<String>>, + pub poll_interval: Option<String>, + pub poll_timeout: Option<String>, + pub batch_size: Option<u32>, + /// Only `json` is accepted today. `arrow_ipc` needs the batch scanner and a different + /// offset-tracking path, so it is rejected rather than quietly downgraded. + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option<SecretString>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap<i32, i64>, + messages_produced: u64, +} + +#[derive(Debug, Clone, Copy)] +enum StartingOffset { + Earliest, + Latest, + Explicit(i64), +} + +impl FromStr for StartingOffset { + type Err = Error; + + fn from_str(value: &str) -> Result<Self, Self::Err> { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => match other.parse::<i64>() { + Ok(offset) if offset >= 0 => Ok(StartingOffset::Explicit(offset)), + _ => Err(Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', \ + 'latest' or a non-negative offset" + ))), + }, + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option<FlussConnection>, + scanner: Option<LogScanner>, + fields: Vec<DataField>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + /// Set when `open()` resolved start offsets that are not on disk yet. The next poll carries + /// them through the ACK handshake even without rows, so a restart before the first row does + /// not resolve `latest` again and skip what was written in between. + start_offsets_unsaved: AtomicBool, +} + +/// `FlussConnection` and `LogScanner` do not implement `Debug`, so the derive is replaced by +/// a hand-written one that reports connection presence instead of client internals. +impl std::fmt::Debug for FlussSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussSource") + .field("id", &self.id) + .field("table_path", &self.table_path) + .field("poll_interval", &self.poll_interval) + .field("poll_timeout", &self.poll_timeout) + .field("include_metadata", &self.include_metadata) + .field("columns", &self.fields.len()) + .field("opened", &self.scanner.is_some()) + .finish_non_exhaustive() + } +} + +impl FlussSource { + pub fn new(id: u32, config: FlussSourceConfig, state: Option<ConnectorState>) -> Self { + let poll_interval = parse_duration( + config.poll_interval.as_deref(), + DEFAULT_POLL_INTERVAL, + "poll_interval", + id, + ); + let poll_timeout = parse_duration( + config.poll_timeout.as_deref(), + DEFAULT_POLL_TIMEOUT, + "poll_timeout", + id, + ); + let include_metadata = config.include_metadata.unwrap_or(false); + let verbose_logging = config.verbose_logging.unwrap_or(false); + let table_path = TablePath::new(config.database.clone(), config.table.clone()); + + let restored_state = state + .and_then(|state| state.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|state| { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Buckets tracked: {}, messages produced: {}", + state.bucket_offsets.len(), + state.messages_produced + ); + }); + + FlussSource { + id, + config, + table_path, + poll_interval, + poll_timeout, + include_metadata, + verbose_logging, + connection: None, + scanner: None, + fields: Vec::new(), + state: Mutex::new(restored_state.unwrap_or_default()), + pending_state: Mutex::new(None), + start_offsets_unsaved: AtomicBool::new(false), + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn client_config(&self) -> Config { + let mut config = Config { + bootstrap_servers: self.config.bootstrap_servers.clone(), + ..Config::default() + }; + if let Some(batch_size) = self.config.batch_size { + config.scanner_log_max_poll_records = batch_size as usize; + } + if let (Some(username), Some(password)) = + (&self.config.sasl_username, &self.config.sasl_password) + { + config.security_protocol = "sasl".to_owned(); + config.security_sasl_mechanism = "PLAIN".to_owned(); + config.security_sasl_username = username.clone(); + config.security_sasl_password = password.expose_secret().to_owned(); + } + config + } + + fn validate_config(&self) -> Result<StartingOffset, Error> { + let table_type = self.config.table_type.as_deref().unwrap_or(LOG_TABLE_TYPE); + if table_type != LOG_TABLE_TYPE { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only table_type '{LOG_TABLE_TYPE}', got '{table_type}'. \ + Primary-key changelog tables are not supported yet" + ))); + } + + let payload_format = self + .config + .payload_format + .as_deref() + .unwrap_or(JSON_PAYLOAD_FORMAT); + if payload_format != JSON_PAYLOAD_FORMAT { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only payload_format '{JSON_PAYLOAD_FORMAT}', got '{payload_format}'" + ))); + } + + // The client accepts a zero limit, which makes every poll come back empty. + if self.config.batch_size == Some(0) { + return Err(Error::InitError(format!( + "batch_size for {CONNECTOR_NAME} must be greater than 0" + ))); + } + // One without the other would connect without SASL and only fail at the server. + if self.config.sasl_username.is_some() != self.config.sasl_password.is_some() { + return Err(Error::InitError(format!( + "sasl_username and sasl_password for {CONNECTOR_NAME} must be set together" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Buckets already present in the restored state keep their offset. Everything else + /// starts at the given default, so a widened bucket count does not rewind buckets + /// that were already consumed. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } + + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets: HashMap<i32, i64> = HashMap::new(); + for (bucket, bucket_records) in records.records_by_buckets() { Review Comment: Sized from `records_by_buckets().len()` now, and written once per bucket from its last record. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1037 @@ +// 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. + +mod mapping; + +use async_trait::async_trait; +use fluss::client::{EARLIEST_OFFSET, FlussConnection, LogScanner}; +use fluss::config::Config; +use fluss::metadata::{DataField, TablePath}; +use fluss::record::ScanRecords; +use fluss::rpc::message::OffsetSpec; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, info, warn}; + +source_connector!(FlussSource); + +const CONNECTOR_NAME: &str = "Apache Fluss source"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(5); +const LOG_TABLE_TYPE: &str = "log"; +const JSON_PAYLOAD_FORMAT: &str = "json"; +const METADATA_BUCKET: &str = "_fluss_bucket"; +const METADATA_OFFSET: &str = "_fluss_offset"; +const METADATA_TIMESTAMP: &str = "_fluss_timestamp"; +const NANOS_PER_MILLI: u64 = 1_000_000; +/// A rewind is local scanner bookkeeping unless the client has to refresh the table's +/// metadata first. An error from `on_batch_result` stops the source, so that refresh gets a +/// few attempts before the failure is reported. +const REWIND_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(1), +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. A primary-key table streams a changelog whose change + /// types need their own mapping onto messages, so the value is validated rather than + /// silently ignored. + pub table_type: Option<String>, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option<String>, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option<Vec<String>>, + pub poll_interval: Option<String>, + pub poll_timeout: Option<String>, + pub batch_size: Option<u32>, + /// Only `json` is accepted today. `arrow_ipc` needs the batch scanner and a different + /// offset-tracking path, so it is rejected rather than quietly downgraded. + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option<SecretString>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap<i32, i64>, + messages_produced: u64, +} + +#[derive(Debug, Clone, Copy)] +enum StartingOffset { + Earliest, + Latest, + Explicit(i64), +} + +impl FromStr for StartingOffset { + type Err = Error; + + fn from_str(value: &str) -> Result<Self, Self::Err> { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => match other.parse::<i64>() { + Ok(offset) if offset >= 0 => Ok(StartingOffset::Explicit(offset)), + _ => Err(Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', \ + 'latest' or a non-negative offset" + ))), + }, + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option<FlussConnection>, + scanner: Option<LogScanner>, + fields: Vec<DataField>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + /// Set when `open()` resolved start offsets that are not on disk yet. The next poll carries + /// them through the ACK handshake even without rows, so a restart before the first row does + /// not resolve `latest` again and skip what was written in between. + start_offsets_unsaved: AtomicBool, +} + +/// `FlussConnection` and `LogScanner` do not implement `Debug`, so the derive is replaced by +/// a hand-written one that reports connection presence instead of client internals. +impl std::fmt::Debug for FlussSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussSource") + .field("id", &self.id) + .field("table_path", &self.table_path) + .field("poll_interval", &self.poll_interval) + .field("poll_timeout", &self.poll_timeout) + .field("include_metadata", &self.include_metadata) + .field("columns", &self.fields.len()) + .field("opened", &self.scanner.is_some()) + .finish_non_exhaustive() + } +} + +impl FlussSource { + pub fn new(id: u32, config: FlussSourceConfig, state: Option<ConnectorState>) -> Self { + let poll_interval = parse_duration( + config.poll_interval.as_deref(), + DEFAULT_POLL_INTERVAL, + "poll_interval", + id, + ); + let poll_timeout = parse_duration( + config.poll_timeout.as_deref(), + DEFAULT_POLL_TIMEOUT, + "poll_timeout", + id, + ); + let include_metadata = config.include_metadata.unwrap_or(false); + let verbose_logging = config.verbose_logging.unwrap_or(false); + let table_path = TablePath::new(config.database.clone(), config.table.clone()); + + let restored_state = state + .and_then(|state| state.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|state| { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Buckets tracked: {}, messages produced: {}", + state.bucket_offsets.len(), + state.messages_produced + ); + }); + + FlussSource { + id, + config, + table_path, + poll_interval, + poll_timeout, + include_metadata, + verbose_logging, + connection: None, + scanner: None, + fields: Vec::new(), + state: Mutex::new(restored_state.unwrap_or_default()), + pending_state: Mutex::new(None), + start_offsets_unsaved: AtomicBool::new(false), + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn client_config(&self) -> Config { + let mut config = Config { + bootstrap_servers: self.config.bootstrap_servers.clone(), + ..Config::default() + }; + if let Some(batch_size) = self.config.batch_size { + config.scanner_log_max_poll_records = batch_size as usize; + } + if let (Some(username), Some(password)) = + (&self.config.sasl_username, &self.config.sasl_password) + { + config.security_protocol = "sasl".to_owned(); + config.security_sasl_mechanism = "PLAIN".to_owned(); + config.security_sasl_username = username.clone(); + config.security_sasl_password = password.expose_secret().to_owned(); + } + config + } + + fn validate_config(&self) -> Result<StartingOffset, Error> { + let table_type = self.config.table_type.as_deref().unwrap_or(LOG_TABLE_TYPE); + if table_type != LOG_TABLE_TYPE { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only table_type '{LOG_TABLE_TYPE}', got '{table_type}'. \ + Primary-key changelog tables are not supported yet" + ))); + } + + let payload_format = self + .config + .payload_format + .as_deref() + .unwrap_or(JSON_PAYLOAD_FORMAT); + if payload_format != JSON_PAYLOAD_FORMAT { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only payload_format '{JSON_PAYLOAD_FORMAT}', got '{payload_format}'" + ))); + } + + // The client accepts a zero limit, which makes every poll come back empty. + if self.config.batch_size == Some(0) { + return Err(Error::InitError(format!( + "batch_size for {CONNECTOR_NAME} must be greater than 0" + ))); + } + // One without the other would connect without SASL and only fail at the server. + if self.config.sasl_username.is_some() != self.config.sasl_password.is_some() { + return Err(Error::InitError(format!( + "sasl_username and sasl_password for {CONNECTOR_NAME} must be set together" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Buckets already present in the restored state keep their offset. Everything else + /// starts at the given default, so a widened bucket count does not rewind buckets + /// that were already consumed. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } + + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets: HashMap<i32, i64> = HashMap::new(); + for (bucket, bucket_records) in records.records_by_buckets() { + let bucket_id = bucket.bucket_id(); + for record in bucket_records { + let row = mapping::row_to_json(record.row(), &self.fields)?; + messages.push(self.build_message( + bucket_id, + record.offset(), + record.timestamp(), + row, + )?); + polled_offsets.insert(bucket_id, record.offset() + 1); + } + } + + let state = self + .stage_batch_state(polled_offsets, messages.len()) + .await?; + Ok((messages, state)) + } + + fn build_message( + &self, + bucket: i32, + offset: i64, + timestamp_millis: i64, + mut record: serde_json::Map<String, Value>, + ) -> Result<ProducedMessage, Error> { + if self.include_metadata { + record.insert(METADATA_BUCKET.to_owned(), Value::from(bucket)); + record.insert(METADATA_OFFSET.to_owned(), Value::from(offset)); + record.insert(METADATA_TIMESTAMP.to_owned(), Value::from(timestamp_millis)); + } + + let payload = simd_json::to_vec(&Value::Object(record)).map_err(|error| { + Error::Serialization(format!( + "failed to serialize Apache Fluss row at bucket {bucket}, offset {offset}: {error}" + )) + })?; + + Ok(ProducedMessage { + id: Some(message_id(bucket, offset)), + headers: None, + checksum: None, + timestamp: None, + origin_timestamp: origin_timestamp_nanos(timestamp_millis), + payload, + }) + } + + /// Stages the state a polled batch would leave behind until the runtime reports its + /// result. A batch without rows changes no offset, so it carries no state unless the start + /// offsets from `open()` still have to reach disk. + async fn stage_batch_state( + &self, + polled_offsets: HashMap<i32, i64>, + produced: usize, + ) -> Result<Option<ConnectorState>, Error> { + if produced == 0 && !self.start_offsets_unsaved.load(Ordering::Acquire) { + return Ok(None); + } + + let mut candidate = self.state.lock().await.clone(); + candidate.bucket_offsets.extend(polled_offsets); + candidate.messages_produced += produced as u64; + let persisted = self.serialize_state(&candidate).ok_or_else(|| { + Error::Serialization(format!( + "failed to serialize state for {CONNECTOR_NAME} connector with ID: {}", + self.id + )) + })?; + *self.pending_state.lock().await = Some(candidate); + Ok(Some(persisted)) + } + + /// The scanner moves past records as soon as it returns them, so a batch that is not + /// acknowledged is only read again once the scanner points back at the committed offsets. + /// A fetch still buffered from the old position is dropped by the scanner's own + /// expected-offset check. + async fn rewind(&self, scanner: &LogScanner) -> Result<(), Error> { + let committed = { self.state.lock().await.bucket_offsets.clone() }; + let context = format!("{CONNECTOR_NAME} connector with ID: {} rewind", self.id); + retry_async( + REWIND_RETRY, + &context, + fluss::error::Error::is_retriable, + || scanner.subscribe_buckets(&committed), + ) + .await + .map_err(|failure| { + Error::Connection(format!( + "failed to rewind the Apache Fluss scanner to the last acknowledged offsets: \ + {failure}" + )) + }) + } +} + +#[async_trait] +impl Source for FlussSource { + async fn open(&mut self) -> Result<(), Error> { + let start = self.validate_config()?; + + let connection = FlussConnection::new(self.client_config()) + .await + .map_err(connection_error)?; + + // The scanner aligns every row to the schema it was created with, so the columns used to + // decode rows come from the same table snapshot rather than from a second lookup that a + // schema change could land between. + let (scanner, fields, bucket_count) = { + let table = connection + .get_table(&self.table_path) + .await + .map_err(connection_error)?; + let table_info = table.get_table_info(); + + if table_info.has_primary_key() { + return Err(Error::InitError(format!( + "table '{}' is a primary-key table. {CONNECTOR_NAME} supports log tables only", + self.table_path + ))); + } + if table_info.is_partitioned() { + return Err(Error::InitError(format!( + "table '{}' is partitioned, which {CONNECTOR_NAME} does not support yet", + self.table_path + ))); + } + + let row_type = match self.config.columns.as_ref() { + Some(columns) => table_info + .get_row_type() + .project_with_field_names(columns) + .map_err(|error| { + Error::InitError(format!( + "invalid columns projection {columns:?} for table '{}': {error}", + self.table_path + )) + })?, + None => table_info.get_row_type().clone(), + }; + mapping::ensure_supported_types(row_type.fields())?; + + let scan = match self.config.columns.as_ref() { + Some(columns) => { + let names: Vec<&str> = columns.iter().map(String::as_str).collect(); + table.new_scan().project_by_name(&names).map_err(|error| { + Error::InitError(format!("failed to project columns: {error}")) + })? + } + None => table.new_scan(), + }; + let scanner = scan.create_log_scanner().map_err(|error| { + Error::InitError(format!("failed to create log scanner: {error}")) + })?; + ( + scanner, + row_type.fields().clone(), + table_info.get_num_buckets(), + ) + }; + + let restored = { self.state.lock().await.bucket_offsets.clone() }; + let offsets = match start { + StartingOffset::Earliest => { + Self::resolve_start_offsets(bucket_count, EARLIEST_OFFSET, &restored) + } + StartingOffset::Explicit(offset) => { + Self::resolve_start_offsets(bucket_count, offset, &restored) + } + StartingOffset::Latest => { + let missing: Vec<i32> = (0..bucket_count) + .filter(|bucket| !restored.contains_key(bucket)) + .collect(); + let mut offsets = restored.clone(); + if !missing.is_empty() { + let admin = connection.get_admin().map_err(connection_error)?; + let tails = admin + .list_offsets(&self.table_path, &missing, OffsetSpec::Latest) + .await + .map_err(connection_error)?; + for bucket in missing { + let tail = tails.get(&bucket).copied().ok_or_else(|| { + Error::InitError(format!( + "Apache Fluss returned no latest offset for bucket {bucket} of table '{}'", + self.table_path + )) + })?; + offsets.insert(bucket, tail); + } + } + offsets + } + }; + scanner + .subscribe_buckets(&offsets) + .await + .map_err(connection_error)?; + + self.start_offsets_unsaved + .store(offsets != restored, Ordering::Release); + { + let mut state = self.state.lock().await; + state.bucket_offsets = offsets; + } + + self.fields = fields; + self.connection = Some(connection); + self.scanner = Some(scanner); Review Comment: Documented under Limitations. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1037 @@ +// 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. + +mod mapping; + +use async_trait::async_trait; +use fluss::client::{EARLIEST_OFFSET, FlussConnection, LogScanner}; +use fluss::config::Config; +use fluss::metadata::{DataField, TablePath}; +use fluss::record::ScanRecords; +use fluss::rpc::message::OffsetSpec; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, info, warn}; + +source_connector!(FlussSource); + +const CONNECTOR_NAME: &str = "Apache Fluss source"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(5); +const LOG_TABLE_TYPE: &str = "log"; +const JSON_PAYLOAD_FORMAT: &str = "json"; +const METADATA_BUCKET: &str = "_fluss_bucket"; +const METADATA_OFFSET: &str = "_fluss_offset"; +const METADATA_TIMESTAMP: &str = "_fluss_timestamp"; +const NANOS_PER_MILLI: u64 = 1_000_000; +/// A rewind is local scanner bookkeeping unless the client has to refresh the table's +/// metadata first. An error from `on_batch_result` stops the source, so that refresh gets a +/// few attempts before the failure is reported. +const REWIND_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(1), +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. A primary-key table streams a changelog whose change + /// types need their own mapping onto messages, so the value is validated rather than + /// silently ignored. + pub table_type: Option<String>, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option<String>, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option<Vec<String>>, + pub poll_interval: Option<String>, + pub poll_timeout: Option<String>, + pub batch_size: Option<u32>, + /// Only `json` is accepted today. `arrow_ipc` needs the batch scanner and a different + /// offset-tracking path, so it is rejected rather than quietly downgraded. + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option<SecretString>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap<i32, i64>, + messages_produced: u64, +} + +#[derive(Debug, Clone, Copy)] +enum StartingOffset { + Earliest, + Latest, + Explicit(i64), +} + +impl FromStr for StartingOffset { + type Err = Error; + + fn from_str(value: &str) -> Result<Self, Self::Err> { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => match other.parse::<i64>() { + Ok(offset) if offset >= 0 => Ok(StartingOffset::Explicit(offset)), + _ => Err(Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', \ + 'latest' or a non-negative offset" + ))), + }, + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option<FlussConnection>, + scanner: Option<LogScanner>, + fields: Vec<DataField>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + /// Set when `open()` resolved start offsets that are not on disk yet. The next poll carries + /// them through the ACK handshake even without rows, so a restart before the first row does + /// not resolve `latest` again and skip what was written in between. + start_offsets_unsaved: AtomicBool, +} + +/// `FlussConnection` and `LogScanner` do not implement `Debug`, so the derive is replaced by +/// a hand-written one that reports connection presence instead of client internals. +impl std::fmt::Debug for FlussSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussSource") + .field("id", &self.id) + .field("table_path", &self.table_path) + .field("poll_interval", &self.poll_interval) + .field("poll_timeout", &self.poll_timeout) + .field("include_metadata", &self.include_metadata) + .field("columns", &self.fields.len()) + .field("opened", &self.scanner.is_some()) + .finish_non_exhaustive() + } +} + +impl FlussSource { + pub fn new(id: u32, config: FlussSourceConfig, state: Option<ConnectorState>) -> Self { + let poll_interval = parse_duration( + config.poll_interval.as_deref(), + DEFAULT_POLL_INTERVAL, + "poll_interval", + id, + ); + let poll_timeout = parse_duration( + config.poll_timeout.as_deref(), + DEFAULT_POLL_TIMEOUT, + "poll_timeout", + id, + ); + let include_metadata = config.include_metadata.unwrap_or(false); + let verbose_logging = config.verbose_logging.unwrap_or(false); + let table_path = TablePath::new(config.database.clone(), config.table.clone()); + + let restored_state = state + .and_then(|state| state.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|state| { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Buckets tracked: {}, messages produced: {}", + state.bucket_offsets.len(), + state.messages_produced + ); + }); + + FlussSource { + id, + config, + table_path, + poll_interval, + poll_timeout, + include_metadata, + verbose_logging, + connection: None, + scanner: None, + fields: Vec::new(), + state: Mutex::new(restored_state.unwrap_or_default()), + pending_state: Mutex::new(None), + start_offsets_unsaved: AtomicBool::new(false), + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn client_config(&self) -> Config { + let mut config = Config { + bootstrap_servers: self.config.bootstrap_servers.clone(), + ..Config::default() + }; + if let Some(batch_size) = self.config.batch_size { + config.scanner_log_max_poll_records = batch_size as usize; + } + if let (Some(username), Some(password)) = + (&self.config.sasl_username, &self.config.sasl_password) + { + config.security_protocol = "sasl".to_owned(); + config.security_sasl_mechanism = "PLAIN".to_owned(); + config.security_sasl_username = username.clone(); + config.security_sasl_password = password.expose_secret().to_owned(); + } + config + } + + fn validate_config(&self) -> Result<StartingOffset, Error> { + let table_type = self.config.table_type.as_deref().unwrap_or(LOG_TABLE_TYPE); + if table_type != LOG_TABLE_TYPE { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only table_type '{LOG_TABLE_TYPE}', got '{table_type}'. \ + Primary-key changelog tables are not supported yet" + ))); + } + + let payload_format = self + .config + .payload_format + .as_deref() + .unwrap_or(JSON_PAYLOAD_FORMAT); + if payload_format != JSON_PAYLOAD_FORMAT { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only payload_format '{JSON_PAYLOAD_FORMAT}', got '{payload_format}'" + ))); + } + + // The client accepts a zero limit, which makes every poll come back empty. + if self.config.batch_size == Some(0) { + return Err(Error::InitError(format!( + "batch_size for {CONNECTOR_NAME} must be greater than 0" + ))); + } + // One without the other would connect without SASL and only fail at the server. + if self.config.sasl_username.is_some() != self.config.sasl_password.is_some() { + return Err(Error::InitError(format!( + "sasl_username and sasl_password for {CONNECTOR_NAME} must be set together" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Buckets already present in the restored state keep their offset. Everything else + /// starts at the given default, so a widened bucket count does not rewind buckets + /// that were already consumed. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } + + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets: HashMap<i32, i64> = HashMap::new(); + for (bucket, bucket_records) in records.records_by_buckets() { + let bucket_id = bucket.bucket_id(); + for record in bucket_records { + let row = mapping::row_to_json(record.row(), &self.fields)?; + messages.push(self.build_message( + bucket_id, + record.offset(), + record.timestamp(), + row, + )?); + polled_offsets.insert(bucket_id, record.offset() + 1); Review Comment: Went with skip and log. The row is dropped with an error naming its bucket and offset, the offsets move past it, and `close()` logs how many rows were skipped, since a plugin can't bump the runtime's error metric. The runtime and the HTTP sink drop messages they can't decode the same way. Escalating wouldn't stop the loop here, because the SDK only logs a poll error and polls again. There's a new integration test with a date chrono can't represent. The branch at :522 now only handles a state serialization failure. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1037 @@ +// 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. + +mod mapping; + +use async_trait::async_trait; +use fluss::client::{EARLIEST_OFFSET, FlussConnection, LogScanner}; +use fluss::config::Config; +use fluss::metadata::{DataField, TablePath}; +use fluss::record::ScanRecords; +use fluss::rpc::message::OffsetSpec; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, info, warn}; + +source_connector!(FlussSource); + +const CONNECTOR_NAME: &str = "Apache Fluss source"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(5); +const LOG_TABLE_TYPE: &str = "log"; +const JSON_PAYLOAD_FORMAT: &str = "json"; +const METADATA_BUCKET: &str = "_fluss_bucket"; +const METADATA_OFFSET: &str = "_fluss_offset"; +const METADATA_TIMESTAMP: &str = "_fluss_timestamp"; +const NANOS_PER_MILLI: u64 = 1_000_000; +/// A rewind is local scanner bookkeeping unless the client has to refresh the table's +/// metadata first. An error from `on_batch_result` stops the source, so that refresh gets a +/// few attempts before the failure is reported. +const REWIND_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(1), +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. A primary-key table streams a changelog whose change + /// types need their own mapping onto messages, so the value is validated rather than + /// silently ignored. + pub table_type: Option<String>, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option<String>, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option<Vec<String>>, + pub poll_interval: Option<String>, + pub poll_timeout: Option<String>, + pub batch_size: Option<u32>, + /// Only `json` is accepted today. `arrow_ipc` needs the batch scanner and a different + /// offset-tracking path, so it is rejected rather than quietly downgraded. + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option<SecretString>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap<i32, i64>, + messages_produced: u64, +} + +#[derive(Debug, Clone, Copy)] +enum StartingOffset { + Earliest, + Latest, + Explicit(i64), +} + +impl FromStr for StartingOffset { + type Err = Error; + + fn from_str(value: &str) -> Result<Self, Self::Err> { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => match other.parse::<i64>() { + Ok(offset) if offset >= 0 => Ok(StartingOffset::Explicit(offset)), + _ => Err(Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', \ + 'latest' or a non-negative offset" + ))), + }, + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option<FlussConnection>, + scanner: Option<LogScanner>, + fields: Vec<DataField>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + /// Set when `open()` resolved start offsets that are not on disk yet. The next poll carries + /// them through the ACK handshake even without rows, so a restart before the first row does + /// not resolve `latest` again and skip what was written in between. + start_offsets_unsaved: AtomicBool, +} + +/// `FlussConnection` and `LogScanner` do not implement `Debug`, so the derive is replaced by +/// a hand-written one that reports connection presence instead of client internals. +impl std::fmt::Debug for FlussSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussSource") + .field("id", &self.id) + .field("table_path", &self.table_path) + .field("poll_interval", &self.poll_interval) + .field("poll_timeout", &self.poll_timeout) + .field("include_metadata", &self.include_metadata) + .field("columns", &self.fields.len()) + .field("opened", &self.scanner.is_some()) + .finish_non_exhaustive() + } +} + +impl FlussSource { + pub fn new(id: u32, config: FlussSourceConfig, state: Option<ConnectorState>) -> Self { + let poll_interval = parse_duration( + config.poll_interval.as_deref(), + DEFAULT_POLL_INTERVAL, + "poll_interval", + id, + ); + let poll_timeout = parse_duration( + config.poll_timeout.as_deref(), + DEFAULT_POLL_TIMEOUT, + "poll_timeout", + id, + ); + let include_metadata = config.include_metadata.unwrap_or(false); + let verbose_logging = config.verbose_logging.unwrap_or(false); + let table_path = TablePath::new(config.database.clone(), config.table.clone()); + + let restored_state = state + .and_then(|state| state.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|state| { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Buckets tracked: {}, messages produced: {}", + state.bucket_offsets.len(), + state.messages_produced + ); + }); + + FlussSource { + id, + config, + table_path, + poll_interval, + poll_timeout, + include_metadata, + verbose_logging, + connection: None, + scanner: None, + fields: Vec::new(), + state: Mutex::new(restored_state.unwrap_or_default()), + pending_state: Mutex::new(None), + start_offsets_unsaved: AtomicBool::new(false), + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn client_config(&self) -> Config { + let mut config = Config { + bootstrap_servers: self.config.bootstrap_servers.clone(), + ..Config::default() + }; + if let Some(batch_size) = self.config.batch_size { + config.scanner_log_max_poll_records = batch_size as usize; + } + if let (Some(username), Some(password)) = + (&self.config.sasl_username, &self.config.sasl_password) + { + config.security_protocol = "sasl".to_owned(); + config.security_sasl_mechanism = "PLAIN".to_owned(); + config.security_sasl_username = username.clone(); + config.security_sasl_password = password.expose_secret().to_owned(); + } + config + } + + fn validate_config(&self) -> Result<StartingOffset, Error> { + let table_type = self.config.table_type.as_deref().unwrap_or(LOG_TABLE_TYPE); + if table_type != LOG_TABLE_TYPE { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only table_type '{LOG_TABLE_TYPE}', got '{table_type}'. \ + Primary-key changelog tables are not supported yet" + ))); + } + + let payload_format = self + .config + .payload_format + .as_deref() + .unwrap_or(JSON_PAYLOAD_FORMAT); + if payload_format != JSON_PAYLOAD_FORMAT { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only payload_format '{JSON_PAYLOAD_FORMAT}', got '{payload_format}'" + ))); + } + + // The client accepts a zero limit, which makes every poll come back empty. + if self.config.batch_size == Some(0) { + return Err(Error::InitError(format!( + "batch_size for {CONNECTOR_NAME} must be greater than 0" + ))); + } + // One without the other would connect without SASL and only fail at the server. + if self.config.sasl_username.is_some() != self.config.sasl_password.is_some() { + return Err(Error::InitError(format!( + "sasl_username and sasl_password for {CONNECTOR_NAME} must be set together" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Buckets already present in the restored state keep their offset. Everything else + /// starts at the given default, so a widened bucket count does not rewind buckets + /// that were already consumed. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } Review Comment: Both paths build the subscription from `0..bucket_count` now, so a saved offset for a bucket the table no longer has is dropped, with a warning naming it. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1037 @@ +// 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. + +mod mapping; + +use async_trait::async_trait; +use fluss::client::{EARLIEST_OFFSET, FlussConnection, LogScanner}; +use fluss::config::Config; +use fluss::metadata::{DataField, TablePath}; +use fluss::record::ScanRecords; +use fluss::rpc::message::OffsetSpec; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, info, warn}; + +source_connector!(FlussSource); + +const CONNECTOR_NAME: &str = "Apache Fluss source"; +const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(5); +const LOG_TABLE_TYPE: &str = "log"; +const JSON_PAYLOAD_FORMAT: &str = "json"; +const METADATA_BUCKET: &str = "_fluss_bucket"; +const METADATA_OFFSET: &str = "_fluss_offset"; +const METADATA_TIMESTAMP: &str = "_fluss_timestamp"; +const NANOS_PER_MILLI: u64 = 1_000_000; +/// A rewind is local scanner bookkeeping unless the client has to refresh the table's +/// metadata first. An error from `on_batch_result` stops the source, so that refresh gets a +/// few attempts before the failure is reported. +const REWIND_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(200), + max_delay: Duration::from_secs(1), +}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. A primary-key table streams a changelog whose change + /// types need their own mapping onto messages, so the value is validated rather than + /// silently ignored. + pub table_type: Option<String>, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option<String>, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option<Vec<String>>, + pub poll_interval: Option<String>, + pub poll_timeout: Option<String>, + pub batch_size: Option<u32>, + /// Only `json` is accepted today. `arrow_ipc` needs the batch scanner and a different + /// offset-tracking path, so it is rejected rather than quietly downgraded. + pub payload_format: Option<String>, + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option<SecretString>, + pub verbose_logging: Option<bool>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap<i32, i64>, + messages_produced: u64, +} + +#[derive(Debug, Clone, Copy)] +enum StartingOffset { + Earliest, + Latest, + Explicit(i64), +} + +impl FromStr for StartingOffset { + type Err = Error; + + fn from_str(value: &str) -> Result<Self, Self::Err> { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => match other.parse::<i64>() { + Ok(offset) if offset >= 0 => Ok(StartingOffset::Explicit(offset)), + _ => Err(Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', \ + 'latest' or a non-negative offset" + ))), + }, + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option<FlussConnection>, + scanner: Option<LogScanner>, + fields: Vec<DataField>, + state: Mutex<State>, + pending_state: Mutex<Option<State>>, + /// Set when `open()` resolved start offsets that are not on disk yet. The next poll carries + /// them through the ACK handshake even without rows, so a restart before the first row does + /// not resolve `latest` again and skip what was written in between. + start_offsets_unsaved: AtomicBool, +} + +/// `FlussConnection` and `LogScanner` do not implement `Debug`, so the derive is replaced by +/// a hand-written one that reports connection presence instead of client internals. +impl std::fmt::Debug for FlussSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FlussSource") + .field("id", &self.id) + .field("table_path", &self.table_path) + .field("poll_interval", &self.poll_interval) + .field("poll_timeout", &self.poll_timeout) + .field("include_metadata", &self.include_metadata) + .field("columns", &self.fields.len()) + .field("opened", &self.scanner.is_some()) + .finish_non_exhaustive() + } +} + +impl FlussSource { + pub fn new(id: u32, config: FlussSourceConfig, state: Option<ConnectorState>) -> Self { + let poll_interval = parse_duration( + config.poll_interval.as_deref(), + DEFAULT_POLL_INTERVAL, + "poll_interval", + id, + ); + let poll_timeout = parse_duration( + config.poll_timeout.as_deref(), + DEFAULT_POLL_TIMEOUT, + "poll_timeout", + id, + ); + let include_metadata = config.include_metadata.unwrap_or(false); + let verbose_logging = config.verbose_logging.unwrap_or(false); + let table_path = TablePath::new(config.database.clone(), config.table.clone()); + + let restored_state = state + .and_then(|state| state.deserialize::<State>(CONNECTOR_NAME, id)) + .inspect(|state| { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Buckets tracked: {}, messages produced: {}", + state.bucket_offsets.len(), + state.messages_produced + ); + }); + + FlussSource { + id, + config, + table_path, + poll_interval, + poll_timeout, + include_metadata, + verbose_logging, + connection: None, + scanner: None, + fields: Vec::new(), + state: Mutex::new(restored_state.unwrap_or_default()), + pending_state: Mutex::new(None), + start_offsets_unsaved: AtomicBool::new(false), + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn client_config(&self) -> Config { + let mut config = Config { + bootstrap_servers: self.config.bootstrap_servers.clone(), + ..Config::default() + }; + if let Some(batch_size) = self.config.batch_size { + config.scanner_log_max_poll_records = batch_size as usize; + } + if let (Some(username), Some(password)) = + (&self.config.sasl_username, &self.config.sasl_password) + { + config.security_protocol = "sasl".to_owned(); + config.security_sasl_mechanism = "PLAIN".to_owned(); + config.security_sasl_username = username.clone(); + config.security_sasl_password = password.expose_secret().to_owned(); + } + config + } + + fn validate_config(&self) -> Result<StartingOffset, Error> { + let table_type = self.config.table_type.as_deref().unwrap_or(LOG_TABLE_TYPE); + if table_type != LOG_TABLE_TYPE { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only table_type '{LOG_TABLE_TYPE}', got '{table_type}'. \ + Primary-key changelog tables are not supported yet" + ))); + } + + let payload_format = self + .config + .payload_format + .as_deref() + .unwrap_or(JSON_PAYLOAD_FORMAT); + if payload_format != JSON_PAYLOAD_FORMAT { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} supports only payload_format '{JSON_PAYLOAD_FORMAT}', got '{payload_format}'" + ))); + } + + // The client accepts a zero limit, which makes every poll come back empty. + if self.config.batch_size == Some(0) { + return Err(Error::InitError(format!( + "batch_size for {CONNECTOR_NAME} must be greater than 0" + ))); + } + // One without the other would connect without SASL and only fail at the server. + if self.config.sasl_username.is_some() != self.config.sasl_password.is_some() { + return Err(Error::InitError(format!( + "sasl_username and sasl_password for {CONNECTOR_NAME} must be set together" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Buckets already present in the restored state keep their offset. Everything else + /// starts at the given default, so a widened bucket count does not rewind buckets + /// that were already consumed. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } + + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets: HashMap<i32, i64> = HashMap::new(); + for (bucket, bucket_records) in records.records_by_buckets() { + let bucket_id = bucket.bucket_id(); + for record in bucket_records { + let row = mapping::row_to_json(record.row(), &self.fields)?; + messages.push(self.build_message( + bucket_id, + record.offset(), + record.timestamp(), + row, + )?); + polled_offsets.insert(bucket_id, record.offset() + 1); + } + } + + let state = self + .stage_batch_state(polled_offsets, messages.len()) + .await?; + Ok((messages, state)) + } + + fn build_message( + &self, + bucket: i32, + offset: i64, + timestamp_millis: i64, + mut record: serde_json::Map<String, Value>, + ) -> Result<ProducedMessage, Error> { + if self.include_metadata { + record.insert(METADATA_BUCKET.to_owned(), Value::from(bucket)); + record.insert(METADATA_OFFSET.to_owned(), Value::from(offset)); + record.insert(METADATA_TIMESTAMP.to_owned(), Value::from(timestamp_millis)); + } + + let payload = simd_json::to_vec(&Value::Object(record)).map_err(|error| { + Error::Serialization(format!( + "failed to serialize Apache Fluss row at bucket {bucket}, offset {offset}: {error}" + )) + })?; Review Comment: `open()` now rejects a column under `_fluss_` while `include_metadata` is on, and the README says the prefix is reserved. I kept the flat layout, since nesting the metadata would change the payload shape. An integration test checks that such a table fails to start. -- 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]
