seokjin0414 commented on code in PR #3799: URL: https://github.com/apache/iggy/pull/3799#discussion_r4161976313
########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1385 @@ +// 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, RowType, 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, HashSet}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, error, 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 METADATA_PREFIX: &str = "_fluss_"; +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. An empty list or + /// a repeated name is rejected. + 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>, + /// Adds the bucket, offset and timestamp of each row under the `_fluss_` prefix, so a column + /// under that prefix is rejected while it is on. + pub include_metadata: Option<bool>, + pub sasl_username: Option<String>, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] Review Comment: Fixed in f8151c1. `Serialize` is gone together with the annotation and the `iggy_common` dependency, so serializing the config no longer compiles. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1385 @@ +// 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, RowType, 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, HashSet}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, error, 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 METADATA_PREFIX: &str = "_fluss_"; +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. An empty list or + /// a repeated name is rejected. + 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>, + /// Adds the bucket, offset and timestamp of each row under the `_fluss_` prefix, so a column + /// under that prefix is rejected while it is on. + 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, + rows_skipped: AtomicU64, +} + +/// `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), + rows_skipped: AtomicU64::new(0), + } + } + + 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> { + for (field, value) in [ + ("bootstrap_servers", &self.config.bootstrap_servers), + ("database", &self.config.database), + ("table", &self.config.table), + ] { + if value.trim().is_empty() { + return Err(Error::InitError(format!( + "{field} for {CONNECTOR_NAME} must not be empty" + ))); + } + } + + 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}'" + ))); + } + + if let Some(columns) = &self.config.columns { + if columns.is_empty() { + return Err(Error::InitError(format!( + "columns for {CONNECTOR_NAME} must name at least one column, or be left out to \ + read every column" + ))); + } + let mut seen = HashSet::with_capacity(columns.len()); + if let Some(duplicate) = columns.iter().find(|column| !seen.insert(column.as_str())) { + return Err(Error::InitError(format!( + "column '{duplicate}' is listed more than once in columns for {CONNECTOR_NAME}" + ))); + } + } + + // 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" + ))); + } + // With neither a pause before the poll nor a server-side wait, an idle table would be + // polled in a tight loop. + if self.poll_interval.is_zero() && self.poll_timeout.is_zero() { + return Err(Error::InitError(format!( + "poll_interval and poll_timeout for {CONNECTOR_NAME} cannot both be zero" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Only the table's current buckets are subscribed. A bucket already present in the restored + /// state keeps its offset and every other one starts at the given default, so a widened + /// bucket count does not rewind buckets that were already consumed, and an offset saved for + /// a bucket the table no longer has is dropped. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + (0..bucket_count) + .map(|bucket| { + ( + bucket, + tracked.get(&bucket).copied().unwrap_or(start_offset), + ) + }) + .collect() + } + + /// Same rule as [`Self::resolve_start_offsets`], with each untracked bucket starting at its + /// current tail. + async fn resolve_latest_offsets( + &self, + connection: &FlussConnection, + bucket_count: i32, + tracked: &HashMap<i32, i64>, + ) -> Result<HashMap<i32, i64>, Error> { + let missing: Vec<i32> = (0..bucket_count) + .filter(|bucket| !tracked.contains_key(bucket)) + .collect(); + let tails = if missing.is_empty() { + HashMap::new() + } else { + let admin = connection.get_admin().map_err(connection_error)?; + admin + .list_offsets(&self.table_path, &missing, OffsetSpec::Latest) + .await + .map_err(connection_error)? + }; + (0..bucket_count) + .map(|bucket| { + let offset = match tracked.get(&bucket) { + Some(offset) => *offset, + None => tails.get(&bucket).copied().ok_or_else(|| { + Error::InitError(format!( + "Apache Fluss returned no latest offset for bucket {bucket} of table '{}'", + self.table_path + )) + })?, + }; + Ok((bucket, offset)) + }) + .collect() + } + + /// A row that cannot be converted fails the same way on every read, so failing the batch + /// would read it again forever. It is dropped and logged with its position instead, and the + /// offsets still move past it. + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let buckets = records.records_by_buckets(); + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets = HashMap::with_capacity(buckets.len()); + let mut skipped = 0u64; + for (bucket, bucket_records) in buckets { + let bucket_id = bucket.bucket_id(); + for record in bucket_records { + let message = mapping::row_to_json(record.row(), &self.fields).and_then(|row| { + self.build_message(bucket_id, record.offset(), record.timestamp(), row) + }); + match message { + Ok(message) => messages.push(message), + Err(error) => { Review Comment: A column read cannot fail transiently here. `ScanRecord::row()` reads a batch that is already decoded in memory (`ColumnarRow` over an `Arc<TypedBatch>`), and in fluss-rs 1.0.0 every accessor error is an `IllegalArgument` for an out-of-bounds index or a type mismatch, which `Error::is_retriable()` reports as not retriable. Failing the batch on it would read the same row again forever. `read_error` also mapped every read failure to `InvalidRecordValue`, so skipping only that variant would skip the same rows. I added the reason to the `build_batch` doc comment in f8151c1. ########## core/connectors/sources/fluss_source/src/mapping.rs: ########## @@ -0,0 +1,355 @@ +// 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 base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64; +use chrono::{DateTime, NaiveDate, NaiveTime, Utc}; +use fluss::metadata::{DataField, DataType}; +use fluss::row::InternalRow; +use iggy_connector_sdk::Error; +use serde_json::{Map, Number, Value}; + +const MILLIS_PER_SECOND: i64 = 1_000; +const NANOS_PER_MILLI: i64 = 1_000_000; + +/// Temporal values are formatted with their full fractional precision, the way the +/// PostgreSQL source formats them. A number of milliseconds would drop the microseconds of +/// the default `TIMESTAMP(6)`. `TIMESTAMP` carries no timezone and is written without one, +/// while `TIMESTAMP_LTZ` is an instant and is written in UTC. +pub(crate) fn row_to_json( + row: &dyn InternalRow, + fields: &[DataField], +) -> Result<Map<String, Value>, Error> { + let mut object = Map::with_capacity(fields.len()); + for (position, field) in fields.iter().enumerate() { + let value = read_field(row, position, field.data_type())?; + object.insert(field.name().to_owned(), value); Review Comment: Fixed in f8151c1. `JsonRow` now writes each row straight into the payload, borrowing column names and string values. The JSON is unchanged: I checked it byte for byte against the old map-based mapping on random rows and schemas. ########## core/connectors/sources/fluss_source/README.md: ########## @@ -0,0 +1,90 @@ +# Apache Fluss source connector + +Reads rows from an [Apache Fluss](https://fluss.apache.org/) log table and publishes them to an Apache Iggy stream as JSON. + +Each Fluss row becomes one Apache Iggy message. Offsets are tracked per bucket and persisted through the runtime state API, so a restart resumes where the previous run stopped. Offsets only advance once the runtime acknowledges a delivered batch; a rejected batch rewinds the scanner to the last acknowledged offsets and is read again. The start position is persisted with the first poll even when it returns no rows, so a restart does not resolve `latest` again past rows written in between. + +## Configuration + +| Field | Required | Default | Description | +| ----- | -------- | ------- | ----------- | +| `bootstrap_servers` | yes | | Coordinator server address, for example `localhost:9123`. | +| `database` | yes | | Apache Fluss database name. | +| `table` | yes | | Apache Fluss table name. | +| `table_type` | no | `log` | Only `log` is accepted. See [Limitations](#limitations). | +| `starting_offset` | no | `earliest` | `earliest`, `latest` (each bucket's tail, resolved at startup), or an explicit non-negative offset. Applies only to buckets absent from the persisted state. An explicit offset is used as is for every such bucket, so it suits a table whose bucket offsets you know, such as one with a single bucket. | +| `columns` | no | all columns | Column projection pushed down to the server. Must name at least one column, and no column twice. | +| `poll_interval` | no | `1s` | Delay before each poll. | +| `poll_timeout` | no | `5s` | How long a single server poll waits for records. It and `poll_interval` cannot both be zero. | +| `batch_size` | no | `500` | Maximum records returned per poll (`scanner.log.max-poll-records`). Must be greater than 0. | +| `payload_format` | no | `json` | Only `json` is accepted. See [Limitations](#limitations). | +| `include_metadata` | no | `false` | Adds `_fluss_bucket`, `_fluss_offset` and `_fluss_timestamp` to each JSON object. While it is on, a column whose name starts with `_fluss_` is rejected at startup, since it would be overwritten. | +| `sasl_username` | no | | Enables SASL/PLAIN together with `sasl_password`. Set both or neither. | +| `sasl_password` | no | | Stored as a secret and redacted from logs and the `/stats` endpoint. | +| `verbose_logging` | no | `false` | Logs per-batch counts at info instead of debug. | + +## Example + +```toml +type = "source" +key = "fluss" +enabled = true +version = 0 +name = "Apache Fluss source" +path = "libiggy_connector_fluss_source" + +[[streams]] +stream = "fluss_events" +topic = "events" +schema = "json" +batch_length = 100 + +[plugin_config] +bootstrap_servers = "localhost:9123" +database = "mydb" +table = "events" +poll_interval = "1s" +batch_size = 500 +``` + +Bring up a local cluster with the [official Docker compose recipe](https://fluss.apache.org/docs/install-deploy/deploying-with-docker/) (ZooKeeper, one coordinator server, one tablet server), create the stream and topic with the Apache Iggy CLI, then start the connectors runtime. + +## Type mapping + +| Apache Fluss type | JSON | +| ----------------- | ---- | +| `BOOLEAN` | boolean | +| `TINYINT`, `SMALLINT`, `INT`, `BIGINT` | number | +| `FLOAT`, `DOUBLE` | number, `null` when not finite | +| `CHAR`, `STRING` | string | +| `DECIMAL` | string, to keep the full precision | +| `DATE` | string, `2024-02-29` | +| `TIME` | string, `12:34:56.789` | +| `TIMESTAMP` | string without a timezone, `2023-11-14 22:13:20.123456` | +| `TIMESTAMP_LTZ` | string in UTC, RFC 3339, `2023-11-14T22:13:20.123456+00:00` | +| `BINARY`, `BYTES` | base64 string | +| `ARRAY`, `MAP`, `ROW` | not supported, rejected at startup | + +Temporal values keep every fractional digit the column holds, so the default `TIMESTAMP(6)` keeps its microseconds. They are formatted the same way the PostgreSQL source formats them: `TIMESTAMP` carries no timezone and is written without one, while `TIMESTAMP_LTZ` is an instant and is written in UTC. + +Rows are decoded with the table schema read when the connector starts. A row written before a column was added carries `null` for it, and a column added after startup stays out of the output until the connector restarts. + +Every message carries an `id` derived from its bucket and offset, so a consumer can spot a record replayed after an at-least-once redelivery (the Apache Iggy server does not deduplicate on it), and an `origin_timestamp` taken from the Fluss record timestamp. Review Comment: Fixed in f8151c1. The README now says the runtime does not pass `origin_timestamp` on and points to `_fluss_timestamp` under `include_metadata`. ########## core/connectors/sources/fluss_source/README.md: ########## @@ -0,0 +1,90 @@ +# Apache Fluss source connector + +Reads rows from an [Apache Fluss](https://fluss.apache.org/) log table and publishes them to an Apache Iggy stream as JSON. + +Each Fluss row becomes one Apache Iggy message. Offsets are tracked per bucket and persisted through the runtime state API, so a restart resumes where the previous run stopped. Offsets only advance once the runtime acknowledges a delivered batch; a rejected batch rewinds the scanner to the last acknowledged offsets and is read again. The start position is persisted with the first poll even when it returns no rows, so a restart does not resolve `latest` again past rows written in between. + +## Configuration + +| Field | Required | Default | Description | +| ----- | -------- | ------- | ----------- | +| `bootstrap_servers` | yes | | Coordinator server address, for example `localhost:9123`. | +| `database` | yes | | Apache Fluss database name. | +| `table` | yes | | Apache Fluss table name. | +| `table_type` | no | `log` | Only `log` is accepted. See [Limitations](#limitations). | +| `starting_offset` | no | `earliest` | `earliest`, `latest` (each bucket's tail, resolved at startup), or an explicit non-negative offset. Applies only to buckets absent from the persisted state. An explicit offset is used as is for every such bucket, so it suits a table whose bucket offsets you know, such as one with a single bucket. | +| `columns` | no | all columns | Column projection pushed down to the server. Must name at least one column, and no column twice. | +| `poll_interval` | no | `1s` | Delay before each poll. | +| `poll_timeout` | no | `5s` | How long a single server poll waits for records. It and `poll_interval` cannot both be zero. | +| `batch_size` | no | `500` | Maximum records returned per poll (`scanner.log.max-poll-records`). Must be greater than 0. | +| `payload_format` | no | `json` | Only `json` is accepted. See [Limitations](#limitations). | +| `include_metadata` | no | `false` | Adds `_fluss_bucket`, `_fluss_offset` and `_fluss_timestamp` to each JSON object. While it is on, a column whose name starts with `_fluss_` is rejected at startup, since it would be overwritten. | +| `sasl_username` | no | | Enables SASL/PLAIN together with `sasl_password`. Set both or neither. | +| `sasl_password` | no | | Stored as a secret and redacted from logs and the `/stats` endpoint. | Review Comment: Fixed in f8151c1. The row now says the runtime logs the raw plugin config at trace level and returns it from its config API, and keeps the `/stats` part. ########## core/connectors/sources/fluss_source/src/lib.rs: ########## @@ -0,0 +1,1385 @@ +// 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, RowType, 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, HashSet}; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::sleep; +use tracing::{debug, error, 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 METADATA_PREFIX: &str = "_fluss_"; +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. An empty list or + /// a repeated name is rejected. + 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>, + /// Adds the bucket, offset and timestamp of each row under the `_fluss_` prefix, so a column + /// under that prefix is rejected while it is on. + 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, + rows_skipped: AtomicU64, +} + +/// `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), + rows_skipped: AtomicU64::new(0), + } + } + + 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> { + for (field, value) in [ + ("bootstrap_servers", &self.config.bootstrap_servers), + ("database", &self.config.database), + ("table", &self.config.table), + ] { + if value.trim().is_empty() { + return Err(Error::InitError(format!( + "{field} for {CONNECTOR_NAME} must not be empty" + ))); + } + } + + 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}'" + ))); + } + + if let Some(columns) = &self.config.columns { + if columns.is_empty() { + return Err(Error::InitError(format!( + "columns for {CONNECTOR_NAME} must name at least one column, or be left out to \ + read every column" + ))); + } + let mut seen = HashSet::with_capacity(columns.len()); + if let Some(duplicate) = columns.iter().find(|column| !seen.insert(column.as_str())) { + return Err(Error::InitError(format!( + "column '{duplicate}' is listed more than once in columns for {CONNECTOR_NAME}" + ))); + } + } + + // 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" + ))); + } + // With neither a pause before the poll nor a server-side wait, an idle table would be + // polled in a tight loop. + if self.poll_interval.is_zero() && self.poll_timeout.is_zero() { + return Err(Error::InitError(format!( + "poll_interval and poll_timeout for {CONNECTOR_NAME} cannot both be zero" + ))); + } + + self.config + .starting_offset + .as_deref() + .unwrap_or("earliest") + .parse() + } + + /// Only the table's current buckets are subscribed. A bucket already present in the restored + /// state keeps its offset and every other one starts at the given default, so a widened + /// bucket count does not rewind buckets that were already consumed, and an offset saved for + /// a bucket the table no longer has is dropped. + fn resolve_start_offsets( + bucket_count: i32, + start_offset: i64, + tracked: &HashMap<i32, i64>, + ) -> HashMap<i32, i64> { + (0..bucket_count) + .map(|bucket| { + ( + bucket, + tracked.get(&bucket).copied().unwrap_or(start_offset), + ) + }) + .collect() + } + + /// Same rule as [`Self::resolve_start_offsets`], with each untracked bucket starting at its + /// current tail. + async fn resolve_latest_offsets( + &self, + connection: &FlussConnection, + bucket_count: i32, + tracked: &HashMap<i32, i64>, + ) -> Result<HashMap<i32, i64>, Error> { + let missing: Vec<i32> = (0..bucket_count) + .filter(|bucket| !tracked.contains_key(bucket)) + .collect(); + let tails = if missing.is_empty() { + HashMap::new() + } else { + let admin = connection.get_admin().map_err(connection_error)?; + admin + .list_offsets(&self.table_path, &missing, OffsetSpec::Latest) + .await + .map_err(connection_error)? + }; + (0..bucket_count) + .map(|bucket| { + let offset = match tracked.get(&bucket) { + Some(offset) => *offset, + None => tails.get(&bucket).copied().ok_or_else(|| { + Error::InitError(format!( + "Apache Fluss returned no latest offset for bucket {bucket} of table '{}'", + self.table_path + )) + })?, + }; + Ok((bucket, offset)) + }) + .collect() + } + + /// A row that cannot be converted fails the same way on every read, so failing the batch + /// would read it again forever. It is dropped and logged with its position instead, and the + /// offsets still move past it. + async fn build_batch( + &self, + records: &ScanRecords, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let buckets = records.records_by_buckets(); + let mut messages = Vec::with_capacity(records.count()); + let mut polled_offsets = HashMap::with_capacity(buckets.len()); + let mut skipped = 0u64; + for (bucket, bucket_records) in buckets { + let bucket_id = bucket.bucket_id(); + for record in bucket_records { + let message = mapping::row_to_json(record.row(), &self.fields).and_then(|row| { + self.build_message(bucket_id, record.offset(), record.timestamp(), row) + }); + match message { + Ok(message) => messages.push(message), + Err(error) => { + skipped += 1; + error!( + "{CONNECTOR_NAME} connector with ID: {} skipped the row at bucket \ + {bucket_id}, offset {}, which it cannot convert: {error}", + self.id, + record.offset() + ); + } + } + } + if let Some(last) = bucket_records.last() { + polled_offsets.insert(bucket_id, last.offset() + 1); + } + } + if skipped > 0 { + self.rows_skipped.fetch_add(skipped, Ordering::Relaxed); Review Comment: Fixed in f8151c1. The count now rides the staged state and commits on ACK, like `messages_produced`, so a row read again after a NACK is counted once. A unit test covers a NACK followed by an ACK of the same rows. ########## core/integration/tests/connectors/fixtures/fluss/container.rs: ########## @@ -0,0 +1,140 @@ +// 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 crate::connectors::fixtures; +use integration::harness::TestBinaryError; +use std::net::TcpListener; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +const FLUSS_IMAGE: &str = "apache/fluss"; +const FLUSS_VERSION: &str = "1.0.0"; +const READY_MESSAGE: &str = "Registered tablet server 0"; + +pub(super) const ENV_SOURCE_BOOTSTRAP_SERVERS: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_BOOTSTRAP_SERVERS"; +pub(super) const ENV_SOURCE_DATABASE: &str = "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_DATABASE"; +pub(super) const ENV_SOURCE_TABLE: &str = "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_TABLE"; +pub(super) const ENV_SOURCE_COLUMNS: &str = "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_COLUMNS"; +pub(super) const ENV_SOURCE_STARTING_OFFSET: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_STARTING_OFFSET"; +pub(super) const ENV_SOURCE_POLL_INTERVAL: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_POLL_INTERVAL"; +pub(super) const ENV_SOURCE_INCLUDE_METADATA: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_PLUGIN_CONFIG_INCLUDE_METADATA"; +pub(super) const ENV_SOURCE_STREAMS_0_STREAM: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_STREAMS_0_STREAM"; +pub(super) const ENV_SOURCE_STREAMS_0_TOPIC: &str = "IGGY_CONNECTORS_SOURCE_FLUSS_STREAMS_0_TOPIC"; +pub(super) const ENV_SOURCE_STREAMS_0_SCHEMA: &str = + "IGGY_CONNECTORS_SOURCE_FLUSS_STREAMS_0_SCHEMA"; +pub(super) const ENV_SOURCE_PATH: &str = "IGGY_CONNECTORS_SOURCE_FLUSS_PATH"; + +/// A whole Fluss cluster (embedded ZooKeeper, coordinator server, tablet server) inside one +/// container. +/// +/// The image ships `local-cluster.sh`, which starts the same three processes, but it rewrites +/// the tablet server's bind port to 0 so it cannot collide with the coordinator. A random port +/// inside the container cannot be published to the host, so the tablet server is given an +/// explicit second port here instead. +/// +/// Both servers advertise `localhost`, which resolves to the coordinator/tablet pair inside the +/// container and to the published ports from the test process. That only holds while the host +/// and container port numbers match, so free host ports are reserved up front and mapped +/// one-to-one rather than letting Docker assign them. +pub(super) struct FlussContainer { + #[allow(dead_code)] + container: ContainerAsync<GenericImage>, + pub(super) bootstrap_servers: String, +} + +impl FlussContainer { + pub(super) async fn start() -> Result<Self, TestBinaryError> { + let coordinator_port = reserve_host_port()?; Review Comment: Fixed in f8151c1. `reserve_host_ports` keeps every listener open until the last port is read. -- 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]
