From 4b98471f6f6b1331b7ba7c3b5dc4718ce5b101ee Mon Sep 17 00:00:00 2001 From: seokjin0414 Date: Sun, 2 Aug 2026 20:37:17 +0900 Subject: [PATCH 1/4] feat(connectors): add Apache Fluss source connector Apache Fluss keeps streams as schema-aware columnar log tables, so feeding one into Apache Iggy so far meant writing a bespoke client. This connector reads a Fluss log table and publishes each row as JSON, tracking offsets per bucket through the runtime state API so a restart resumes where the previous run stopped. Buckets already present in the restored state keep their offset, which keeps a widened bucket count from rewinding buckets that were already consumed. Scope follows the released fluss-rs 0.1.0. Primary-key changelog scanning is absent from that release, and arrow_ipc payloads need a separate scanner with its own offset-tracking path, so both are rejected at startup instead of being silently downgraded. Resolving `starting_offset = "latest"` needs an offset spec type the client does not export, so that value is rejected with the reason. Column projection is pushed down to the server when `columns` is set, and temporal values keep their Fluss-native integer form because Fluss carries no timezone that a formatted string could honour. Signed-off-by: seokjin0414 --- .../connectors/fluss_source.toml | 45 ++ core/connectors/sources/README.md | 1 + .../sources/fluss_source/Cargo.toml | 52 ++ .../connectors/sources/fluss_source/README.md | 86 +++ .../sources/fluss_source/config.toml | 48 ++ .../sources/fluss_source/src/lib.rs | 709 ++++++++++++++++++ .../sources/fluss_source/src/mapping.rs | 224 ++++++ 7 files changed, 1165 insertions(+) create mode 100644 core/connectors/runtime/example_config/connectors/fluss_source.toml create mode 100644 core/connectors/sources/fluss_source/Cargo.toml create mode 100644 core/connectors/sources/fluss_source/README.md create mode 100644 core/connectors/sources/fluss_source/config.toml create mode 100644 core/connectors/sources/fluss_source/src/lib.rs create mode 100644 core/connectors/sources/fluss_source/src/mapping.rs diff --git a/core/connectors/runtime/example_config/connectors/fluss_source.toml b/core/connectors/runtime/example_config/connectors/fluss_source.toml new file mode 100644 index 0000000000..9af4a7c387 --- /dev/null +++ b/core/connectors/runtime/example_config/connectors/fluss_source.toml @@ -0,0 +1,45 @@ +# 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. + +type = "source" +key = "fluss" +enabled = true +version = 0 +name = "Apache Fluss source" +path = "/target/release/libiggy_connector_fluss_source" +plugin_config_format = "toml" +verbose = false + +[[streams]] +stream = "events" +topic = "fluss_events" +schema = "json" +batch_length = 100 +linger_time = "5ms" + +[plugin_config] +bootstrap_servers = "localhost:9123" +database = "mydb" +table = "events" +table_type = "log" +starting_offset = "earliest" +poll_interval = "1s" +poll_timeout = "5s" +batch_size = 500 +payload_format = "json" +include_metadata = false +verbose_logging = false diff --git a/core/connectors/sources/README.md b/core/connectors/sources/README.md index 34989aef00..e365bbb476 100644 --- a/core/connectors/sources/README.md +++ b/core/connectors/sources/README.md @@ -9,6 +9,7 @@ Source connectors are responsible for ingesting data from external sources into | Source | Description | | ------ | ----------- | | **elasticsearch_source** | Polls documents from Elasticsearch indices with timestamp-based tracking | +| **fluss_source** | Reads rows from Apache Fluss log tables with per-bucket offset tracking and column projection | | **influxdb_source** | Polls InfluxDB with cursor-based timestamp tracking; supports V2 (Flux, annotated CSV) and V3 (SQL, JSONL) | | **postgres_source** | Reads rows from PostgreSQL tables with multiple strategies: delete after read, mark as processed, or timestamp tracking | | **random_source** | Generates random test messages (useful for testing and development) | diff --git a/core/connectors/sources/fluss_source/Cargo.toml b/core/connectors/sources/fluss_source/Cargo.toml new file mode 100644 index 0000000000..712f5fdfd0 --- /dev/null +++ b/core/connectors/sources/fluss_source/Cargo.toml @@ -0,0 +1,52 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "iggy_connector_fluss_source" +version = "0.5.0-edge.2" +description = "Iggy Apache Fluss source connector for producing messages from Fluss log tables" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "fluss", "source"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org" +documentation = "https://iggy.apache.org/docs" +repository = "https://github.com/apache/iggy" +readme = "../../README.md" +publish = false + +[package.metadata.cargo-machete] +ignored = ["dashmap"] + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +base64 = { workspace = true } +dashmap = { workspace = true } +fluss-rs = { workspace = true } +humantime = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +rmp-serde = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +simd-json = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } diff --git a/core/connectors/sources/fluss_source/README.md b/core/connectors/sources/fluss_source/README.md new file mode 100644 index 0000000000..eb27d96773 --- /dev/null +++ b/core/connectors/sources/fluss_source/README.md @@ -0,0 +1,86 @@ +# 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. + +## 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 numeric offset. Applies only to buckets absent from the persisted state. | +| `columns` | no | all columns | Column projection pushed down to the server. | +| `poll_interval` | no | `1s` | Delay before each poll. | +| `poll_timeout` | no | `5s` | How long a single server poll waits for records. | +| `batch_size` | no | client default | Maximum records returned per poll (`scanner.log.max-poll-records`). | +| `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. | +| `sasl_username` | no | | Enables SASL/PLAIN together with `sasl_password`. | +| `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` | number, days since the Unix epoch | +| `TIME` | number, milliseconds since midnight | +| `TIMESTAMP`, `TIMESTAMP_LTZ` | number, milliseconds since the Unix epoch | +| `BINARY`, `BYTES` | base64 string | +| `ARRAY`, `MAP`, `ROW` | not supported, rejected at startup | + +Temporal values keep their Fluss-native integer representation rather than being formatted, because Fluss does not carry a timezone that a formatted string could honour. + +Every message carries an `id` derived from its bucket and offset, so Apache Iggy can deduplicate after an at-least-once replay, and an `origin_timestamp` taken from the Fluss record timestamp. + +## Limitations + +These follow from the released `fluss-rs` 0.1.0 client rather than from the connector. + +- **Primary-key tables are not supported.** Changelog scanning is not in the released client, so `table_type = "primary_key"` is rejected at startup. +- **`payload_format = "arrow_ipc"` is not implemented yet.** The client does expose an Arrow `RecordBatch` scanner, but it uses a different offset-tracking path, so it is left for a follow-up. +- **Partitioned tables are not supported.** They are detected and rejected at startup. + +## Build and test + +```bash +cargo build --release -p iggy_connector_fluss_source +cargo test -p iggy_connector_fluss_source +``` + +Building this crate requires a system `protoc`, because `fluss-rs` compiles its protocol definitions in a build script. diff --git a/core/connectors/sources/fluss_source/config.toml b/core/connectors/sources/fluss_source/config.toml new file mode 100644 index 0000000000..499b476834 --- /dev/null +++ b/core/connectors/sources/fluss_source/config.toml @@ -0,0 +1,48 @@ +# 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. + +type = "source" +key = "fluss" +enabled = true +version = 0 +name = "Apache Fluss source" +path = "../../target/release/libiggy_connector_fluss_source" +verbose = false + +[[streams]] +stream = "fluss_events" +topic = "events" +schema = "json" +batch_length = 100 + +[plugin_config] +bootstrap_servers = "localhost:9123" +database = "mydb" +table = "events" +table_type = "log" +starting_offset = "earliest" +poll_interval = "1s" +poll_timeout = "5s" +batch_size = 500 +payload_format = "json" +include_metadata = false +verbose_logging = false +# Read a subset of columns. The projection is pushed down to Apache Fluss. +# columns = ["id", "payload"] +# SASL/PLAIN credentials, when the cluster requires them. +# sasl_username = "user" +# sasl_password = "secret" diff --git a/core/connectors/sources/fluss_source/src/lib.rs b/core/connectors/sources/fluss_source/src/lib.rs new file mode 100644 index 0000000000..c58d3ccee6 --- /dev/null +++ b/core/connectors/sources/fluss_source/src/lib.rs @@ -0,0 +1,709 @@ +// 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::rpc::message::OffsetSpec; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, source_connector, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::str::FromStr; +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; + +#[derive(Debug, Serialize, Deserialize)] +pub struct FlussSourceConfig { + pub bootstrap_servers: String, + pub database: String, + pub table: String, + /// Only `log` is accepted today. Primary-key changelog scanning is not in the released + /// `fluss-rs`, so the value is validated rather than silently ignored. + pub table_type: Option, + /// `earliest` (default), `latest`, or an explicit numeric offset applied to every bucket. + pub starting_offset: Option, + /// Column projection pushed down to the server. Omit to read every column. + pub columns: Option>, + pub poll_interval: Option, + pub poll_timeout: Option, + pub batch_size: Option, + /// 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, + pub include_metadata: Option, + pub sasl_username: Option, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] + pub sasl_password: Option, + pub verbose_logging: Option, +} + +#[derive(Debug, Serialize, Deserialize, Default)] +struct State { + /// Next offset to read per bucket. Absent buckets fall back to the configured start. + bucket_offsets: HashMap, + 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 { + match value { + "earliest" => Ok(StartingOffset::Earliest), + "latest" => Ok(StartingOffset::Latest), + other => other.parse::().map(StartingOffset::Explicit).map_err(|_| { + Error::InitError(format!( + "invalid starting_offset '{other}' for {CONNECTOR_NAME}, expected 'earliest', 'latest' or a number" + )) + }), + } + } +} + +pub struct FlussSource { + id: u32, + config: FlussSourceConfig, + table_path: TablePath, + poll_interval: Duration, + poll_timeout: Duration, + include_metadata: bool, + verbose_logging: bool, + connection: Option, + scanner: Option, + fields: Vec, + state: Mutex, +} + +/// `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) -> 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::(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()), + } + } + + fn serialize_state(&self, state: &State) -> Option { + 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 { + 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 scanning is not available in the released fluss-rs" + ))); + } + + 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}'" + ))); + } + + 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, + ) -> HashMap { + let mut offsets = tracked.clone(); + for bucket in 0..bucket_count { + offsets.entry(bucket).or_insert(start_offset); + } + offsets + } + + fn build_message( + &self, + bucket: i32, + offset: i64, + timestamp_millis: i64, + mut record: serde_json::Map, + ) -> Result { + 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, + }) + } +} + +#[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)?; + + let admin = connection.get_admin().map_err(connection_error)?; + let table_info = admin + .get_table_info(&self.table_path) + .await + .map_err(connection_error)?; + + 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 bucket_count = table_info.get_num_buckets(); + let tracked = { self.state.lock().await.bucket_offsets.clone() }; + let offsets = match start { + StartingOffset::Earliest => { + Self::resolve_start_offsets(bucket_count, EARLIEST_OFFSET, &tracked) + } + StartingOffset::Explicit(offset) => { + Self::resolve_start_offsets(bucket_count, offset, &tracked) + } + StartingOffset::Latest => { + let missing: Vec = (0..bucket_count) + .filter(|bucket| !tracked.contains_key(bucket)) + .collect(); + let mut offsets = tracked.clone(); + if !missing.is_empty() { + 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 + } + }; + + let scanner = { + let table = connection + .get_table(&self.table_path) + .await + .map_err(connection_error)?; + 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(), + }; + scan.create_log_scanner().map_err(|error| { + Error::InitError(format!("failed to create log scanner: {error}")) + })? + }; + scanner + .subscribe_buckets(&offsets) + .await + .map_err(connection_error)?; + + { + let mut state = self.state.lock().await; + state.bucket_offsets = offsets; + } + + self.fields = row_type.fields().clone(); + self.connection = Some(connection); + self.scanner = Some(scanner); + + info!( + "Opened {CONNECTOR_NAME} connector with ID: {}, table: {}, buckets: {bucket_count}, \ + columns: {}, poll interval: {:?}", + self.id, + self.table_path, + self.fields.len(), + self.poll_interval + ); + Ok(()) + } + + async fn poll(&self) -> Result { + sleep(self.poll_interval).await; + + let Some(scanner) = self.scanner.as_ref() else { + return Err(Error::InitError(format!( + "{CONNECTOR_NAME} connector with ID: {} polled before it was opened", + self.id + ))); + }; + + let records = scanner + .poll(self.poll_timeout) + .await + .map_err(|error| Error::Connection(format!("failed to poll Apache Fluss: {error}")))?; + + let mut messages = Vec::with_capacity(records.count()); + let mut latest_offsets: HashMap = 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, + )?); + latest_offsets.insert(bucket_id, record.offset() + 1); + } + } + + let produced = messages.len(); + let persisted_state = { + let mut state = self.state.lock().await; + state.bucket_offsets.extend(latest_offsets); + state.messages_produced += produced as u64; + self.serialize_state(&state) + }; + + if produced > 0 { + if self.verbose_logging { + info!( + "{CONNECTOR_NAME} connector with ID: {} produced {produced} messages from table: {}", + self.id, self.table_path + ); + } else { + debug!( + "{CONNECTOR_NAME} connector with ID: {} produced {produced} messages from table: {}", + self.id, self.table_path + ); + } + } + + Ok(ProducedMessages { + schema: Schema::Json, + messages, + state: persisted_state, + }) + } + + async fn close(&mut self) -> Result<(), Error> { + // fluss-rs 0.1.0 exposes no explicit connection shutdown, so dropping is the only + // way to release the client. + self.scanner = None; + self.connection = None; + let state = self.state.lock().await; + info!( + "Closed {CONNECTOR_NAME} connector with ID: {}, total messages produced: {}", + self.id, state.messages_produced + ); + Ok(()) + } +} + +/// Buckets are per-table and offsets are per-bucket, so the pair is unique within a table and +/// stable across restarts. Apache Iggy can dedupe on it after an at-least-once replay. +fn message_id(bucket: i32, offset: i64) -> u128 { + ((bucket as u32 as u128) << 64) | (offset as u64 as u128) +} + +fn origin_timestamp_nanos(timestamp_millis: i64) -> Option { + u64::try_from(timestamp_millis) + .ok() + .and_then(|millis| millis.checked_mul(NANOS_PER_MILLI)) +} + +/// `new()` cannot fail (the FFI macro fixes its signature), so an unparsable duration falls +/// back to the default. It is logged at warn so a typo does not silently change the cadence. +fn parse_duration(value: Option<&str>, default: Duration, field: &str, id: u32) -> Duration { + let Some(raw) = value else { + return default; + }; + match humantime::Duration::from_str(raw) { + Ok(duration) => *duration, + Err(error) => { + warn!( + "Invalid {field} '{raw}' for {CONNECTOR_NAME} connector with ID: {id}, \ + falling back to {default:?}. {error}" + ); + default + } + } +} + +fn connection_error(error: fluss::error::Error) -> Error { + Error::Connection(format!("Apache Fluss client failure: {error}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_config() -> FlussSourceConfig { + FlussSourceConfig { + bootstrap_servers: "localhost:9123".to_owned(), + database: "analytics".to_owned(), + table: "events".to_owned(), + table_type: None, + starting_offset: None, + columns: None, + poll_interval: Some("100ms".to_owned()), + poll_timeout: Some("1s".to_owned()), + batch_size: Some(500), + payload_format: None, + include_metadata: None, + sasl_username: None, + sasl_password: None, + verbose_logging: None, + } + } + + fn state_with(offsets: &[(i32, i64)], produced: u64) -> State { + State { + bucket_offsets: offsets.iter().copied().collect(), + messages_produced: produced, + } + } + + #[test] + fn given_persisted_state_should_restore_bucket_offsets() { + let serialized = rmp_serde::to_vec(&state_with(&[(0, 42), (1, 7)], 500)) + .expect("Failed to serialize state"); + + let source = FlussSource::new(1, test_config(), Some(ConnectorState(serialized))); + + let runtime = tokio::runtime::Runtime::new().expect("Failed to build runtime"); + runtime.block_on(async { + let restored = source.state.lock().await; + assert_eq!(restored.messages_produced, 500); + assert_eq!(restored.bucket_offsets.get(&0), Some(&42)); + assert_eq!(restored.bucket_offsets.get(&1), Some(&7)); + }); + } + + #[test] + fn given_no_state_should_start_fresh() { + let source = FlussSource::new(1, test_config(), None); + + let runtime = tokio::runtime::Runtime::new().expect("Failed to build runtime"); + runtime.block_on(async { + let state = source.state.lock().await; + assert_eq!(state.messages_produced, 0); + assert!(state.bucket_offsets.is_empty()); + }); + } + + #[test] + fn given_invalid_state_should_start_fresh() { + let invalid = ConnectorState(b"not valid msgpack".to_vec()); + + let source = FlussSource::new(1, test_config(), Some(invalid)); + + let runtime = tokio::runtime::Runtime::new().expect("Failed to build runtime"); + runtime.block_on(async { + let state = source.state.lock().await; + assert_eq!(state.messages_produced, 0); + assert!(state.bucket_offsets.is_empty()); + }); + } + + #[test] + fn state_should_be_serializable_and_deserializable() { + let original = state_with(&[(0, 100), (3, 250)], 1000); + + let serialized = rmp_serde::to_vec(&original).expect("Failed to serialize"); + let deserialized: State = + rmp_serde::from_slice(&serialized).expect("Failed to deserialize"); + + assert_eq!(original.messages_produced, deserialized.messages_produced); + assert_eq!(original.bucket_offsets, deserialized.bucket_offsets); + } + + #[test] + fn given_default_config_should_accept_log_table_and_earliest_offset() { + let source = FlussSource::new(1, test_config(), None); + + let start = source + .validate_config() + .expect("Default config should be valid"); + + assert!(matches!(start, StartingOffset::Earliest)); + } + + #[test] + fn given_primary_key_table_type_should_be_rejected() { + let mut config = test_config(); + config.table_type = Some("primary_key".to_owned()); + let source = FlussSource::new(1, config, None); + + let error = source + .validate_config() + .expect_err("Primary key tables are not supported yet"); + + assert!(matches!(error, Error::InitError(message) if message.contains("primary_key"))); + } + + #[test] + fn given_arrow_ipc_payload_format_should_be_rejected() { + let mut config = test_config(); + config.payload_format = Some("arrow_ipc".to_owned()); + let source = FlussSource::new(1, config, None); + + let error = source + .validate_config() + .expect_err("arrow_ipc is not supported yet"); + + assert!(matches!(error, Error::InitError(message) if message.contains("arrow_ipc"))); + } + + #[test] + fn given_latest_starting_offset_should_be_parsed() { + let mut config = test_config(); + config.starting_offset = Some("latest".to_owned()); + let source = FlussSource::new(1, config, None); + + let start = source.validate_config().expect("latest should be accepted"); + + assert!(matches!(start, StartingOffset::Latest)); + } + + #[test] + fn given_explicit_starting_offset_should_be_parsed() { + let mut config = test_config(); + config.starting_offset = Some("128".to_owned()); + let source = FlussSource::new(1, config, None); + + let start = source + .validate_config() + .expect("Explicit offset should parse"); + + assert!(matches!(start, StartingOffset::Explicit(128))); + } + + #[test] + fn given_unparsable_starting_offset_should_be_rejected() { + let mut config = test_config(); + config.starting_offset = Some("beginning".to_owned()); + let source = FlussSource::new(1, config, None); + + assert!(source.validate_config().is_err()); + } + + #[test] + fn given_tracked_buckets_should_keep_their_offsets_and_fill_the_rest() { + let tracked = HashMap::from([(0, 42)]); + + let offsets = FlussSource::resolve_start_offsets(3, EARLIEST_OFFSET, &tracked); + + assert_eq!(offsets.len(), 3); + assert_eq!(offsets[&0], 42); + assert_eq!(offsets[&1], EARLIEST_OFFSET); + assert_eq!(offsets[&2], EARLIEST_OFFSET); + } + + #[test] + fn given_explicit_start_should_apply_to_untracked_buckets_only() { + let tracked = HashMap::from([(1, 900)]); + + let offsets = FlussSource::resolve_start_offsets(2, 50, &tracked); + + assert_eq!(offsets[&0], 50); + assert_eq!(offsets[&1], 900); + } + + #[test] + fn message_id_should_be_unique_per_bucket_and_offset() { + assert_ne!(message_id(0, 1), message_id(1, 0)); + assert_ne!(message_id(0, 1), message_id(0, 2)); + assert_eq!(message_id(2, 5), message_id(2, 5)); + } + + #[test] + fn origin_timestamp_should_convert_milliseconds_to_nanoseconds() { + assert_eq!( + origin_timestamp_nanos(1_785_655_133_842), + Some(1_785_655_133_842_000_000) + ); + assert_eq!(origin_timestamp_nanos(0), Some(0)); + assert_eq!(origin_timestamp_nanos(-1), None); + } + + #[test] + fn given_invalid_duration_should_fall_back_to_default() { + assert_eq!( + parse_duration(Some("nonsense"), DEFAULT_POLL_INTERVAL, "poll_interval", 1), + DEFAULT_POLL_INTERVAL + ); + assert_eq!( + parse_duration(None, DEFAULT_POLL_TIMEOUT, "poll_timeout", 1), + DEFAULT_POLL_TIMEOUT + ); + assert_eq!( + parse_duration(Some("250ms"), DEFAULT_POLL_INTERVAL, "poll_interval", 1), + Duration::from_millis(250) + ); + } +} diff --git a/core/connectors/sources/fluss_source/src/mapping.rs b/core/connectors/sources/fluss_source/src/mapping.rs new file mode 100644 index 0000000000..53c5b08cbb --- /dev/null +++ b/core/connectors/sources/fluss_source/src/mapping.rs @@ -0,0 +1,224 @@ +// 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 fluss::metadata::{DataField, DataType}; +use fluss::row::InternalRow; +use iggy_connector_sdk::Error; +use serde_json::{Map, Number, Value}; + +/// Temporal values are emitted as their Fluss-native integer representation rather than +/// formatted strings: `Date` as days since the Unix epoch, `Time` as milliseconds since +/// midnight, and both timestamp kinds as milliseconds since the epoch. Formatting would +/// force a timezone policy on downstream consumers that Fluss itself does not carry. +pub(crate) fn row_to_json( + row: &dyn InternalRow, + fields: &[DataField], +) -> Result, 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); + } + Ok(object) +} + +/// Rejects column types with no JSON representation before the first poll, so a table with +/// an unsupported column fails at startup instead of once per batch. +pub(crate) fn ensure_supported_types(fields: &[DataField]) -> Result<(), Error> { + for field in fields { + if !is_supported(field.data_type()) { + return Err(Error::SchemaMismatch(format!( + "column '{}' has type {:?}, which the Apache Fluss source cannot map to JSON", + field.name(), + field.data_type() + ))); + } + } + Ok(()) +} + +fn is_supported(data_type: &DataType) -> bool { + !matches!( + data_type, + DataType::Array(_) | DataType::Map(_) | DataType::Row(_) + ) +} + +fn read_field( + row: &dyn InternalRow, + position: usize, + data_type: &DataType, +) -> Result { + if row.is_null_at(position).map_err(read_error)? { + return Ok(Value::Null); + } + + let value = match data_type { + DataType::Boolean(_) => Value::Bool(row.get_boolean(position).map_err(read_error)?), + DataType::TinyInt(_) => Value::from(row.get_byte(position).map_err(read_error)?), + DataType::SmallInt(_) => Value::from(row.get_short(position).map_err(read_error)?), + DataType::Int(_) => Value::from(row.get_int(position).map_err(read_error)?), + DataType::BigInt(_) => Value::from(row.get_long(position).map_err(read_error)?), + DataType::Float(_) => float_value(f64::from(row.get_float(position).map_err(read_error)?)), + DataType::Double(_) => float_value(row.get_double(position).map_err(read_error)?), + DataType::Char(inner) => Value::String( + row.get_char(position, inner.length() as usize) + .map_err(read_error)? + .to_owned(), + ), + DataType::String(_) => { + Value::String(row.get_string(position).map_err(read_error)?.to_owned()) + } + DataType::Decimal(inner) => { + let decimal = row + .get_decimal(position, inner.precision() as usize, inner.scale() as usize) + .map_err(read_error)?; + Value::String(decimal.to_big_decimal().to_string()) + } + DataType::Date(_) => Value::from(row.get_date(position).map_err(read_error)?.get_inner()), + DataType::Time(_) => Value::from(row.get_time(position).map_err(read_error)?.get_inner()), + DataType::Timestamp(inner) => Value::from( + row.get_timestamp_ntz(position, inner.precision()) + .map_err(read_error)? + .get_millisecond(), + ), + DataType::TimestampLTz(inner) => Value::from( + row.get_timestamp_ltz(position, inner.precision()) + .map_err(read_error)? + .get_epoch_millisecond(), + ), + DataType::Bytes(_) => { + Value::String(BASE64.encode(row.get_bytes(position).map_err(read_error)?)) + } + DataType::Binary(inner) => Value::String( + BASE64.encode( + row.get_binary(position, inner.length()) + .map_err(read_error)?, + ), + ), + DataType::Array(_) | DataType::Map(_) | DataType::Row(_) => { + return Err(Error::SchemaMismatch(format!( + "nested type {data_type:?} is not supported by the Apache Fluss source" + ))); + } + }; + Ok(value) +} + +/// JSON has no encoding for NaN or infinity, so those collapse to null rather than +/// failing the whole batch over one degenerate float. +fn float_value(value: f64) -> Value { + Number::from_f64(value).map_or(Value::Null, Value::Number) +} + +fn read_error(error: fluss::error::Error) -> Error { + Error::InvalidRecordValue(format!("failed to read Apache Fluss column: {error}")) +} + +#[cfg(test)] +mod tests { + use super::*; + use fluss::metadata::DataTypes; + use fluss::row::GenericRow; + + fn field(name: &str, data_type: DataType) -> DataField { + DataField::new(name, data_type, None) + } + + #[test] + fn given_scalar_columns_when_mapped_should_produce_json_object() { + let fields = vec![ + field("id", DataTypes::int()), + field("name", DataTypes::string()), + field("active", DataTypes::boolean()), + field("ratio", DataTypes::double()), + field("total", DataTypes::bigint()), + ]; + let mut row = GenericRow::new(5); + row.set_field(0, 7i32); + row.set_field(1, "alice"); + row.set_field(2, true); + row.set_field(3, 1.5f64); + row.set_field(4, 90i64); + + let object = row_to_json(&row, &fields).expect("Failed to map row"); + + assert_eq!(object["id"], Value::from(7)); + assert_eq!(object["name"], Value::from("alice")); + assert_eq!(object["active"], Value::from(true)); + assert_eq!(object["ratio"], Value::from(1.5)); + assert_eq!(object["total"], Value::from(90)); + } + + #[test] + fn given_unset_column_when_mapped_should_produce_null() { + let fields = vec![ + field("id", DataTypes::int()), + field("name", DataTypes::string()), + ]; + let mut row = GenericRow::new(2); + row.set_field(0, 1i32); + + let object = row_to_json(&row, &fields).expect("Failed to map row"); + + assert_eq!(object["id"], Value::from(1)); + assert_eq!(object["name"], Value::Null); + } + + #[test] + fn given_binary_column_when_mapped_should_produce_base64() { + let fields = vec![field("blob", DataTypes::bytes())]; + let mut row = GenericRow::new(1); + row.set_field(0, [1u8, 2, 3].as_slice()); + + let object = row_to_json(&row, &fields).expect("Failed to map row"); + + assert_eq!(object["blob"], Value::from(BASE64.encode([1u8, 2, 3]))); + } + + #[test] + fn given_scalar_columns_when_validated_should_be_accepted() { + let fields = vec![ + field("a", DataTypes::string()), + field("b", DataTypes::timestamp()), + field("c", DataTypes::decimal(10, 2)), + ]; + + assert!(ensure_supported_types(&fields).is_ok()); + } + + #[test] + fn given_nested_column_when_validated_should_be_rejected() { + let fields = vec![ + field("id", DataTypes::int()), + field("tags", DataTypes::array(DataTypes::string())), + ]; + + let error = ensure_supported_types(&fields).expect_err("Nested column should be rejected"); + + assert!(matches!(error, Error::SchemaMismatch(message) if message.contains("tags"))); + } + + #[test] + fn given_non_finite_float_should_map_to_null() { + assert_eq!(float_value(f64::NAN), Value::Null); + assert_eq!(float_value(f64::INFINITY), Value::Null); + assert_eq!(float_value(2.5), Value::from(2.5)); + } +} From 08c6761f7fdb3eee1f5e19a70398cb6da8d6d4fb Mon Sep 17 00:00:00 2001 From: seokjin0414 Date: Sun, 2 Aug 2026 20:37:27 +0900 Subject: [PATCH 2/4] test(integration): cover the Apache Fluss source end to end Exercises the whole path rather than the connector in isolation: rows are appended to a real Fluss log table, and the test asserts on what comes back out of the Apache Iggy topic, including the offsets and bucket the connector attached as metadata. The cluster runs as a single container. The image ships local-cluster.sh, which starts the same embedded ZooKeeper, coordinator server and tablet server, but it rewrites the tablet server's bind port to 0 so it cannot collide with the coordinator, and a random port inside the container cannot be published to the host. Giving the tablet server an explicit second port instead keeps both reachable, and lets both servers advertise localhost, which then resolves the same way inside the container and from the test process. The table is created during fixture setup because the connectors runtime starts before the test body and the connector resolves the table schema while opening. Writes retry because bucket leadership is assigned shortly after the tablet server registers, so the first attempts can still be rejected. Signed-off-by: seokjin0414 --- core/integration/Cargo.toml | 1 + .../connectors/fixtures/fluss/container.rs | 137 +++++++++++++ .../tests/connectors/fixtures/fluss/mod.rs | 21 ++ .../tests/connectors/fixtures/fluss/source.rs | 192 ++++++++++++++++++ .../tests/connectors/fixtures/mod.rs | 2 + .../tests/connectors/fluss/fluss_source.rs | 112 ++++++++++ .../integration/tests/connectors/fluss/mod.rs | 22 ++ .../tests/connectors/fluss/source.toml | 20 ++ core/integration/tests/connectors/mod.rs | 1 + 9 files changed, 508 insertions(+) create mode 100644 core/integration/tests/connectors/fixtures/fluss/container.rs create mode 100644 core/integration/tests/connectors/fixtures/fluss/mod.rs create mode 100644 core/integration/tests/connectors/fixtures/fluss/source.rs create mode 100644 core/integration/tests/connectors/fluss/fluss_source.rs create mode 100644 core/integration/tests/connectors/fluss/mod.rs create mode 100644 core/integration/tests/connectors/fluss/source.toml diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index bb7bdc51b2..e6ab831747 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -52,6 +52,7 @@ ctor = { workspace = true } deltalake = { workspace = true } dtor = { workspace = true } figment = { workspace = true } +fluss-rs = { workspace = true } futures = { workspace = true } harness_derive = { workspace = true } humantime = { workspace = true } diff --git a/core/integration/tests/connectors/fixtures/fluss/container.rs b/core/integration/tests/connectors/fixtures/fluss/container.rs new file mode 100644 index 0000000000..cbbc4bfc3d --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/container.rs @@ -0,0 +1,137 @@ +// 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 = "0.9.1-incubating"; +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_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, + pub(super) bootstrap_servers: String, +} + +impl FlussContainer { + pub(super) async fn start() -> Result { + let coordinator_port = reserve_host_port()?; + let tablet_port = reserve_host_port()?; + + let container = GenericImage::new(FLUSS_IMAGE, FLUSS_VERSION) + .with_wait_for(WaitFor::message_on_stdout(READY_MESSAGE)) + .with_entrypoint("/bin/bash") + .with_container_name(fixtures::unique_container_name("fluss")) + .with_mapped_port(coordinator_port, coordinator_port.tcp()) + .with_mapped_port(tablet_port, tablet_port.tcp()) + .with_env_var("FLUSS_PROPERTIES", server_properties(coordinator_port)) + .with_cmd(["-c", &startup_script(tablet_port)]) + .start() + .await + .map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "FlussContainer".to_string(), + message: format!("Failed to start container: {error}"), + })?; + + Ok(Self { + container, + bootstrap_servers: format!("localhost:{coordinator_port}"), + }) + } +} + +/// The coordinator settings land in `server.yaml`. The tablet server inherits them and +/// overrides only its listeners on the command line. +fn server_properties(coordinator_port: u16) -> String { + format!( + "zookeeper.address: localhost:2181\n\ + bind.listeners: CLIENT://0.0.0.0:{coordinator_port}\n\ + advertised.listeners: CLIENT://localhost:{coordinator_port}\n\ + internal.listener.name: CLIENT\n\ + default.bucket.number: 1\n\ + default.replication.factor: 1\n\ + data.dir: /tmp/fluss/data\n\ + remote.data.dir: /tmp/fluss/remote-data\n\ + tablet-server.id: 0\n" + ) +} + +/// `/docker-entrypoint.sh true` only runs the image's configuration step, which appends +/// `FLUSS_PROPERTIES` to `server.yaml`. The tablet server runs in the foreground so the +/// container stays alive with it. +fn startup_script(tablet_port: u16) -> String { + format!( + "set -e\n\ + /docker-entrypoint.sh true\n\ + /opt/fluss/bin/fluss-daemon.sh start zookeeper /opt/fluss/conf/zookeeper.properties\n\ + /opt/fluss/bin/coordinator-server.sh start\n\ + exec /opt/fluss/bin/tablet-server.sh start-foreground \ + -Dbind.listeners=CLIENT://0.0.0.0:{tablet_port} \ + -Dadvertised.listeners=CLIENT://localhost:{tablet_port}\n" + ) +} + +/// Binds port 0, reads back what the kernel picked, then releases it. The port is only +/// reserved by convention until the container claims it, which is the same trade every +/// fixture that needs a known port ahead of time makes. +fn reserve_host_port() -> Result { + let listener = + TcpListener::bind("127.0.0.1:0").map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "FlussContainer".to_string(), + message: format!("Failed to reserve a host port: {error}"), + })?; + let port = listener + .local_addr() + .map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "FlussContainer".to_string(), + message: format!("Failed to read the reserved host port: {error}"), + })? + .port(); + Ok(port) +} diff --git a/core/integration/tests/connectors/fixtures/fluss/mod.rs b/core/integration/tests/connectors/fixtures/fluss/mod.rs new file mode 100644 index 0000000000..5be251664a --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/mod.rs @@ -0,0 +1,21 @@ +// 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 container; +mod source; + +pub use source::FlussSourceFixture; diff --git a/core/integration/tests/connectors/fixtures/fluss/source.rs b/core/integration/tests/connectors/fixtures/fluss/source.rs new file mode 100644 index 0000000000..b3b63f1fb5 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/source.rs @@ -0,0 +1,192 @@ +// 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 super::container::{ + ENV_SOURCE_BOOTSTRAP_SERVERS, ENV_SOURCE_DATABASE, ENV_SOURCE_INCLUDE_METADATA, + ENV_SOURCE_PATH, ENV_SOURCE_POLL_INTERVAL, ENV_SOURCE_STREAMS_0_SCHEMA, + ENV_SOURCE_STREAMS_0_STREAM, ENV_SOURCE_STREAMS_0_TOPIC, ENV_SOURCE_TABLE, FlussContainer, +}; +use async_trait::async_trait; +use fluss::client::FlussConnection; +use fluss::config::Config; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::GenericRow; +use integration::harness::seeds; +use integration::harness::{TestBinaryError, TestFixture}; +use std::collections::HashMap; +use std::time::Duration; +use tokio::time::sleep; + +const DATABASE: &str = "iggy_test"; +const TABLE: &str = "events"; +const POLL_INTERVAL: &str = "100ms"; +const READY_ATTEMPTS: usize = 20; +const READY_RETRY_DELAY: Duration = Duration::from_millis(500); + +/// Fluss log table read by the source connector under test. +pub struct FlussSourceFixture { + container: FlussContainer, +} + +impl FlussSourceFixture { + /// Appends one row per payload and flushes, so the rows are readable once this returns. + /// + /// Bucket leadership is assigned shortly after the tablet server registers, so the first + /// writes can still be rejected with `NotLeaderOrFollower`. The client retries internally + /// but gives up before leadership settles, hence the retry here. + pub async fn append_rows(&self, payloads: &[String]) -> Result<(), TestBinaryError> { + let mut last_error = None; + for _ in 0..READY_ATTEMPTS { + match self.try_append_rows(payloads).await { + Ok(()) => return Ok(()), + Err(error) => last_error = Some(error), + } + sleep(READY_RETRY_DELAY).await; + } + Err(last_error.unwrap_or_else(|| TestBinaryError::FixtureSetup { + fixture_type: "FlussSourceFixture".to_string(), + message: "Failed to append rows".to_string(), + })) + } + + /// Creates the database and an append-only log table with an `id` and a `payload` column. + /// + /// Runs during `setup()`, before the harness starts the connectors runtime, because the + /// source connector resolves the table schema in `open()` and fails initialization when + /// the table is missing. + async fn create_table_when_ready(&self) -> Result<(), TestBinaryError> { + let mut last_error = None; + for _ in 0..READY_ATTEMPTS { + match self.try_create_table().await { + Ok(()) => return Ok(()), + Err(error) => last_error = Some(error), + } + sleep(READY_RETRY_DELAY).await; + } + Err(last_error.unwrap_or_else(|| TestBinaryError::FixtureSetup { + fixture_type: "FlussSourceFixture".to_string(), + message: "Failed to create table".to_string(), + })) + } + + async fn try_create_table(&self) -> Result<(), TestBinaryError> { + let connection = self.connect().await?; + let admin = connection.get_admin().map_err(|error| self.error(error))?; + admin + .create_database(DATABASE, None, true) + .await + .map_err(|error| self.error(error))?; + + let schema = Schema::builder() + .column("id", DataTypes::int()) + .column("payload", DataTypes::string()) + .build() + .map_err(|error| self.error(error))?; + let descriptor = TableDescriptor::builder() + .schema(schema) + .build() + .map_err(|error| self.error(error))?; + admin + .create_table(&Self::table_path(), &descriptor, true) + .await + .map_err(|error| self.error(error))?; + Ok(()) + } + + async fn try_append_rows(&self, payloads: &[String]) -> Result<(), TestBinaryError> { + let connection = self.connect().await?; + let table = connection + .get_table(&Self::table_path()) + .await + .map_err(|error| self.error(error))?; + let writer = table + .new_append() + .map_err(|error| self.error(error))? + .create_writer() + .map_err(|error| self.error(error))?; + + for (index, payload) in payloads.iter().enumerate() { + let mut row = GenericRow::new(2); + row.set_field(0, index as i32); + row.set_field(1, payload.as_str()); + writer.append(&row).map_err(|error| self.error(error))?; + } + writer.flush().await.map_err(|error| self.error(error))?; + Ok(()) + } + + async fn connect(&self) -> Result { + let config = Config { + bootstrap_servers: self.container.bootstrap_servers.clone(), + ..Config::default() + }; + FlussConnection::new(config) + .await + .map_err(|error| self.error(error)) + } + + fn table_path() -> TablePath { + TablePath::new(DATABASE, TABLE) + } + + fn error(&self, error: fluss::error::Error) -> TestBinaryError { + TestBinaryError::FixtureSetup { + fixture_type: "FlussSourceFixture".to_string(), + message: format!("Apache Fluss client failure: {error}"), + } + } +} + +#[async_trait] +impl TestFixture for FlussSourceFixture { + async fn setup() -> Result { + let fixture = Self { + container: FlussContainer::start().await?, + }; + fixture.create_table_when_ready().await?; + Ok(fixture) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([ + ( + ENV_SOURCE_BOOTSTRAP_SERVERS.to_string(), + self.container.bootstrap_servers.clone(), + ), + (ENV_SOURCE_DATABASE.to_string(), DATABASE.to_string()), + (ENV_SOURCE_TABLE.to_string(), TABLE.to_string()), + ( + ENV_SOURCE_POLL_INTERVAL.to_string(), + POLL_INTERVAL.to_string(), + ), + (ENV_SOURCE_INCLUDE_METADATA.to_string(), "true".to_string()), + ( + ENV_SOURCE_STREAMS_0_STREAM.to_string(), + seeds::names::STREAM.to_string(), + ), + ( + ENV_SOURCE_STREAMS_0_TOPIC.to_string(), + seeds::names::TOPIC.to_string(), + ), + (ENV_SOURCE_STREAMS_0_SCHEMA.to_string(), "json".to_string()), + ( + ENV_SOURCE_PATH.to_string(), + "../../target/debug/libiggy_connector_fluss_source".to_string(), + ), + ]) + } +} diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index e4992d6785..f356787e3e 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -21,6 +21,7 @@ mod clickhouse; mod delta; mod doris; mod elasticsearch; +mod fluss; mod http; mod iceberg; mod influxdb; @@ -57,6 +58,7 @@ pub use doris::{ DorisSinkPreCreatedFixture, }; pub use elasticsearch::{ElasticsearchSinkFixture, ElasticsearchSourcePreCreatedFixture}; +pub use fluss::FlussSourceFixture; pub use http::{ HttpSinkIndividualFixture, HttpSinkJsonArrayFixture, HttpSinkMultiTopicFixture, HttpSinkNdjsonFixture, HttpSinkNoMetadataFixture, HttpSinkRawFixture, diff --git a/core/integration/tests/connectors/fluss/fluss_source.rs b/core/integration/tests/connectors/fluss/fluss_source.rs new file mode 100644 index 0000000000..b9e51a414c --- /dev/null +++ b/core/integration/tests/connectors/fluss/fluss_source.rs @@ -0,0 +1,112 @@ +// 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 super::{POLL_ATTEMPTS, POLL_INTERVAL_MS, TEST_ROW_COUNT}; +use crate::connectors::fixtures::FlussSourceFixture; +use iggy_common::MessageClient; +use iggy_common::{Consumer, Identifier, PollingStrategy}; +use integration::harness::seeds; +use integration::iggy_harness; +use serde::Deserialize; +use std::time::Duration; +use tokio::time::sleep; + +#[derive(Debug, Deserialize)] +struct FlussRecord { + id: i32, + payload: String, + #[serde(rename = "_fluss_bucket")] + bucket: i32, + #[serde(rename = "_fluss_offset")] + offset: i64, + #[serde(rename = "_fluss_timestamp")] + timestamp: i64, +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/source.toml")), + seed = seeds::connector_stream +)] +async fn log_table_rows_are_produced_to_iggy(harness: &TestHarness, fixture: FlussSourceFixture) { + let client = harness.root_client().await.unwrap(); + + let payloads: Vec = (0..TEST_ROW_COUNT) + .map(|index| format!("fluss-payload-{index}")) + .collect(); + fixture + .append_rows(&payloads) + .await + .expect("Failed to append rows"); + + let received = poll_records(&client).await; + + assert!( + received.len() >= TEST_ROW_COUNT, + "Expected at least {TEST_ROW_COUNT} messages, got {}", + received.len() + ); + + for (index, record) in received.iter().take(TEST_ROW_COUNT).enumerate() { + assert_eq!(record.id, index as i32, "Column `id` mismatch at {index}"); + assert_eq!( + record.payload, payloads[index], + "Column `payload` mismatch at {index}" + ); + assert_eq!(record.bucket, 0, "Bucket mismatch at {index}"); + assert_eq!( + record.offset, index as i64, + "Fluss offset should be preserved and sequential at {index}" + ); + assert!( + record.timestamp > 0, + "Fluss timestamp should be populated at {index}" + ); + } +} + +async fn poll_records(client: &iggy::prelude::IggyClient) -> Vec { + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + let consumer_id: Identifier = "fluss_test_consumer".try_into().unwrap(); + + let mut received: Vec = Vec::new(); + for _ in 0..POLL_ATTEMPTS { + if let Ok(polled) = client + .poll_messages( + &stream_id, + &topic_id, + None, + &Consumer::new(consumer_id.clone()), + &PollingStrategy::next(), + 10, + true, + ) + .await + { + for message in polled.messages { + if let Ok(record) = serde_json::from_slice(&message.payload) { + received.push(record); + } + } + if received.len() >= TEST_ROW_COUNT { + break; + } + } + sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; + } + received +} diff --git a/core/integration/tests/connectors/fluss/mod.rs b/core/integration/tests/connectors/fluss/mod.rs new file mode 100644 index 0000000000..462ac01d36 --- /dev/null +++ b/core/integration/tests/connectors/fluss/mod.rs @@ -0,0 +1,22 @@ +// 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 fluss_source; + +const TEST_ROW_COUNT: usize = 10; +const POLL_ATTEMPTS: usize = 30; +const POLL_INTERVAL_MS: u64 = 500; diff --git a/core/integration/tests/connectors/fluss/source.toml b/core/integration/tests/connectors/fluss/source.toml new file mode 100644 index 0000000000..57cb8a6b57 --- /dev/null +++ b/core/integration/tests/connectors/fluss/source.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "../connectors/sources/fluss_source" diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index 08b794fe8b..f394c77867 100644 --- a/core/integration/tests/connectors/mod.rs +++ b/core/integration/tests/connectors/mod.rs @@ -21,6 +21,7 @@ mod delta; mod doris; mod elasticsearch; mod fixtures; +mod fluss; mod http; mod http_config_provider; mod iceberg; From 4365771018c49668e2b09c3576f90c931a4b486e Mon Sep 17 00:00:00 2001 From: seokjin0414 Date: Sun, 2 Aug 2026 20:37:42 +0900 Subject: [PATCH 3/4] build: register the Apache Fluss source in the workspace Registers the connector as a regular workspace member so it inherits the shared dependency versions and stays inside cargo sort, the version bump script and the DAG-based test scoping, the way every other connector does. The alternative is the exclude list, which would keep the workspace build untouched but costs the workspace dependency inheritance, so every dependency would be pinned locally and the crate would fall outside that tooling. Signed-off-by: seokjin0414 --- Cargo.lock | 388 +++++++++++++++++++++++++++++++++++++++++++++++++++-- Cargo.toml | 2 + 2 files changed, 379 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7e44d56d6f..e99712251f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -75,7 +75,7 @@ dependencies = [ "derive_more", "encoding_rs", "flate2", - "foldhash", + "foldhash 0.2.0", "futures-core", "h2 0.3.27", "http 0.2.12", @@ -189,7 +189,7 @@ dependencies = [ "cookie", "derive_more", "encoding_rs", - "foldhash", + "foldhash 0.2.0", "futures-core", "futures-util", "impl-more", @@ -515,6 +515,27 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "arrow" +version = "57.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3bd47f2a6ddc39244bd722a27ee5da66c03369d087b9e024eafdb03e98b98ea7" +dependencies = [ + "arrow-arith 57.3.1", + "arrow-array 57.3.1", + "arrow-buffer 57.3.1", + "arrow-cast 57.3.1", + "arrow-csv 57.3.1", + "arrow-data 57.3.1", + "arrow-ipc 57.3.1", + "arrow-json 57.3.1", + "arrow-ord 57.3.1", + "arrow-row 57.3.1", + "arrow-schema 57.3.1", + "arrow-select 57.3.1", + "arrow-string 57.3.1", +] + [[package]] name = "arrow" version = "58.3.0" @@ -525,12 +546,12 @@ dependencies = [ "arrow-array 58.3.0", "arrow-buffer 58.3.0", "arrow-cast 58.3.0", - "arrow-csv", + "arrow-csv 58.3.0", "arrow-data 58.3.0", "arrow-ipc 58.3.0", "arrow-json 58.3.0", "arrow-ord 58.3.0", - "arrow-row", + "arrow-row 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", "arrow-string 58.3.0", @@ -667,6 +688,21 @@ dependencies = [ "ryu", ] +[[package]] +name = "arrow-csv" +version = "57.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27ddb80a4848e03b1655af496d5ac2563a779e5742fcb48f2ca2e089c9cd2197" +dependencies = [ + "arrow-array 57.3.1", + "arrow-cast 57.3.1", + "arrow-schema 57.3.1", + "chrono", + "csv", + "csv-core", + "regex", +] + [[package]] name = "arrow-csv" version = "58.3.0" @@ -720,6 +756,8 @@ dependencies = [ "arrow-schema 57.3.1", "arrow-select 57.3.1", "flatbuffers", + "lz4_flex 0.12.2", + "zstd", ] [[package]] @@ -811,6 +849,19 @@ dependencies = [ "arrow-select 58.3.0", ] +[[package]] +name = "arrow-row" +version = "57.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a931b520a2a5e22033e01a6f2486b4cdc26f9106b759abeebc320f125e94d7" +dependencies = [ + "arrow-array 57.3.1", + "arrow-buffer 57.3.1", + "arrow-data 57.3.1", + "arrow-schema 57.3.1", + "half", +] + [[package]] name = "arrow-row" version = "58.3.0" @@ -2432,7 +2483,7 @@ version = "0.22.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2235eb320cd7178862a32dd111bd0c0f71a368e393add4914c50129add478eab" dependencies = [ - "arrow", + "arrow 58.3.0", "buoyant_kernel_derive", "bytes", "chrono", @@ -3971,6 +4022,17 @@ dependencies = [ "thiserror 2.0.19", ] +[[package]] +name = "delegate" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "780eb241654bf097afb00fc5f054a09b687dad862e485fdcf8399bb056565370" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "deltalake" version = "0.32.4" @@ -4032,7 +4094,7 @@ version = "0.32.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4588e95ff3b2ccdba56d9ec262bd3467c0593000f729402528706f62be8be1ca" dependencies = [ - "arrow", + "arrow 58.3.0", "arrow-arith 58.3.0", "arrow-array 58.3.0", "arrow-buffer 58.3.0", @@ -4040,7 +4102,7 @@ dependencies = [ "arrow-ipc 58.3.0", "arrow-json 58.3.0", "arrow-ord 58.3.0", - "arrow-row", + "arrow-row 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", "async-trait", @@ -4883,7 +4945,7 @@ version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef975e30683b2d965054bb0a836f8973857c4ebf6acf274fe46617cd285060d8" dependencies = [ - "foldhash", + "foldhash 0.2.0", "libm", "portable-atomic", "siphasher", @@ -5000,6 +5062,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + [[package]] name = "flatbuffers" version = "25.12.19" @@ -5048,12 +5116,58 @@ dependencies = [ "spin", ] +[[package]] +name = "fluss-rs" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fdc8c0e3bf3460948c4156fb733f00718932c95be16888b984336cc49f225fa" +dependencies = [ + "arrow 57.3.1", + "arrow-schema 57.3.1", + "bigdecimal", + "bitvec", + "byteorder", + "bytes", + "clap", + "crc32c", + "dashmap", + "delegate", + "futures", + "jiff", + "linked-hash-map", + "log", + "opendal", + "ordered-float 5.3.0", + "parking_lot", + "parse-display 0.10.0", + "prost", + "prost-build", + "rand 0.9.5", + "scopeguard", + "serde", + "serde_json", + "snafu", + "strum 0.26.3", + "strum_macros 0.26.4", + "tempfile", + "thiserror 1.0.69", + "tokio", + "url", + "uuid", +] + [[package]] name = "fnv" version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "foldhash" version = "0.2.0" @@ -5959,6 +6073,15 @@ version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash 0.1.5", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -5967,7 +6090,7 @@ checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.2.0", ] [[package]] @@ -6972,6 +7095,26 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_fluss_source" +version = "0.5.0-edge.2" +dependencies = [ + "async-trait", + "base64", + "dashmap", + "fluss-rs", + "humantime", + "iggy_common", + "iggy_connector_sdk", + "rmp-serde", + "secrecy", + "serde", + "serde_json", + "simd-json", + "tokio", + "tracing", +] + [[package]] name = "iggy_connector_http_sink" version = "0.5.0-edge.2" @@ -7461,6 +7604,7 @@ dependencies = [ "deltalake", "dtor 1.0.5", "figment", + "fluss-rs", "futures", "harness_derive", "humantime", @@ -7608,10 +7752,12 @@ dependencies = [ "jiff-core", "jiff-static", "jiff-tzdb-platform", + "js-sys", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "wasm-bindgen", "windows-link 0.2.1", ] @@ -8122,6 +8268,9 @@ name = "log" version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +dependencies = [ + "value-bag", +] [[package]] name = "logos" @@ -8718,6 +8867,12 @@ dependencies = [ "pxfm", ] +[[package]] +name = "multimap" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" + [[package]] name = "murmur3" version = "0.5.2" @@ -9353,6 +9508,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", + "rand 0.8.7", + "serde", +] + [[package]] name = "ordered-multimap" version = "0.7.3" @@ -9520,7 +9686,18 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "914a1c2265c98e2446911282c6ac86d8524f495792c38c5bd884f80499c7538a" dependencies = [ - "parse-display-derive", + "parse-display-derive 0.9.1", + "regex", + "regex-syntax", +] + +[[package]] +name = "parse-display" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "287d8d3ebdce117b8539f59411e4ed9ec226e0a4153c7f55495c6070d68e6f72" +dependencies = [ + "parse-display-derive 0.10.0", "regex", "regex-syntax", ] @@ -9539,6 +9716,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "parse-display-derive" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fc048687be30d79502dea2f623d052f3a074012c6eac41726b7ab17213616b1" +dependencies = [ + "proc-macro2", + "quote", + "regex", + "regex-syntax", + "structmeta", + "syn 2.0.119", +] + [[package]] name = "partitions" version = "0.1.0" @@ -9734,6 +9925,17 @@ dependencies = [ "pest", ] +[[package]] +name = "petgraph" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455" +dependencies = [ + "fixedbitset", + "hashbrown 0.15.5", + "indexmap 2.14.0", +] + [[package]] name = "phf" version = "0.12.1" @@ -10176,6 +10378,25 @@ dependencies = [ "prost-derive", ] +[[package]] +name = "prost-build" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" +dependencies = [ + "heck", + "itertools 0.14.0", + "log", + "multimap", + "petgraph", + "prettyplease", + "prost", + "prost-types", + "regex", + "syn 2.0.119", + "tempfile", +] + [[package]] name = "prost-derive" version = "0.14.4" @@ -10457,6 +10678,7 @@ dependencies = [ "libc", "rand_chacha 0.3.1", "rand_core 0.6.4", + "serde", ] [[package]] @@ -10507,6 +10729,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ "getrandom 0.2.17", + "serde", ] [[package]] @@ -11687,6 +11910,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "serde_fmt" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e497af288b3b95d067a23a4f749f2861121ffcb2f6d8379310dcda040c345ed" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_json" version = "1.0.151" @@ -12636,6 +12868,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "strum" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" + [[package]] name = "strum" version = "0.27.2" @@ -12654,6 +12892,19 @@ dependencies = [ "strum_macros 0.28.0", ] +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn 2.0.119", +] + [[package]] name = "strum_macros" version = "0.27.2" @@ -12684,6 +12935,85 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "sval" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba3a370f3cd0422964fd9a19b6516f048527738e6ec50b1b0ff79b460b468390" + +[[package]] +name = "sval_buffer" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "111e6e2dab25782acb41b281263f5f60d6f56a2ba0b3b076187ad23a1552544d" +dependencies = [ + "sval", + "sval_ref", + "zerocopy", +] + +[[package]] +name = "sval_dynamic" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "590ac4ebd299740eac453162302a494e6d5b3be243feab8bf07d0d4331b6b2b3" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_fmt" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b964d6d917267355d7f343c515b908586e5b4aeeada328738f703452f9412d6" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_json" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dea35e4d6b997acc7dc4786ab782f6a6c5000efe4ae93a2db89cf350775fe5fe" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_nested" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a73195ebd4b3e5866db2e1b75f3f383657ad5abc600c646d69b4f064fee89154" +dependencies = [ + "sval", + "sval_buffer", + "sval_ref", +] + +[[package]] +name = "sval_ref" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ef28df9d586bfd15115286651337cb5645f8c203160d22d2d1a20cf13c428c" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_serde" +version = "2.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d72fbe6537e88878f10237bde71ae4a1b65b2bf54bdf274abc566fc696334c92" +dependencies = [ + "serde_core", + "sval", + "sval_nested", +] + [[package]] name = "svgtypes" version = "0.15.3" @@ -12987,7 +13317,7 @@ dependencies = [ "itertools 0.14.0", "log", "memchr", - "parse-display", + "parse-display 0.9.1", "pin-project-lite", "reqwest 0.13.4", "serde", @@ -14129,6 +14459,42 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "value-bag" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" +dependencies = [ + "value-bag-serde1", + "value-bag-sval2", +] + +[[package]] +name = "value-bag-serde1" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "417d6197dd0ee696783d6be4276ac6ea74b985e00024c85ccfb37aff4f2bed82" +dependencies = [ + "erased-serde", + "serde_core", + "serde_fmt", +] + +[[package]] +name = "value-bag-sval2" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c61f7251ecde2c9ed431bbe0659853e7991753447447bbf1ae59d8b31c578d4e" +dependencies = [ + "sval", + "sval_buffer", + "sval_dynamic", + "sval_fmt", + "sval_json", + "sval_ref", + "sval_serde", +] + [[package]] name = "value-trait" version = "0.12.2" diff --git a/Cargo.toml b/Cargo.toml index 31e37bacf1..3cacb24bfc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,6 +48,7 @@ members = [ "core/connectors/sinks/stdout_sink", "core/connectors/sinks/surrealdb_sink", "core/connectors/sources/elasticsearch_source", + "core/connectors/sources/fluss_source", "core/connectors/sources/influxdb_source", "core/connectors/sources/postgres_source", "core/connectors/sources/random_source", @@ -173,6 +174,7 @@ figment = { version = "0.10.19", features = ["toml", "env"] } file-operation = "0.8.28" flatbuffers = "25.12.19" flume = "0.12.0" +fluss-rs = "0.1.0" fs2 = "0.4.3" futures = "0.3.33" futures-core = { version = "0.3.33", default-features = false } From d575c8f850069023b773d0f327a8832c6341b486 Mon Sep 17 00:00:00 2001 From: seokjin0414 Date: Sun, 2 Aug 2026 20:37:42 +0900 Subject: [PATCH 4/4] ci: install protoc for the Apache Fluss connector build The Fluss client compiles its protocol definitions in a build script, so protoc has to be on the PATH for anything that builds the workspace. It is not present on the runner images and prost-build no longer vendors one, so there is nothing to fall back on. The shared Rust setup action already installs system dependencies and is used by every workflow that compiles Rust, so one entry there covers all of them. Signed-off-by: seokjin0414 --- .../actions/utils/setup-rust-with-cache/action.yml | 7 ++++++- CONTRIBUTING.md | 12 ++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/.github/actions/utils/setup-rust-with-cache/action.yml b/.github/actions/utils/setup-rust-with-cache/action.yml index 6c3bf16fb4..282084fe7d 100644 --- a/.github/actions/utils/setup-rust-with-cache/action.yml +++ b/.github/actions/utils/setup-rust-with-cache/action.yml @@ -66,7 +66,9 @@ runs: if: runner.os == 'Linux' && inputs.install-system-dependencies == 'true' run: | sudo apt-get update - sudo apt-get install -y libhwloc-dev pkg-config libudev-dev + # protobuf-compiler: fluss-rs compiles its protocol definitions in a build + # script, so the Apache Fluss connector needs protoc at build time. + sudo apt-get install -y libhwloc-dev pkg-config libudev-dev protobuf-compiler shell: bash - name: Install system dependencies (macOS) @@ -78,6 +80,9 @@ runs: curl -fsSL https://raw.githubusercontent.com/Homebrew/homebrew-core/bb1e23f8e5eacf4d31acd489f6079c8a53ebd690/Formula/h/hwloc.rb \ -o "$(brew --repository iggy/local-hwloc)/Formula/hwloc.rb" brew install iggy/local-hwloc/hwloc + # protobuf-compiler equivalent: fluss-rs compiles its protocol definitions in + # a build script, so the Apache Fluss connector needs protoc at build time. + brew install protobuf shell: bash - name: Setup Rust toolchain diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 8297d919dd..d6069bf21d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -38,6 +38,18 @@ for review. One PR = one thing. Bug fix, refactor, feature - separate PRs. Mixed PRs will be closed. +### System Dependencies + +Building the workspace needs `protoc` on the `PATH`, because the Apache Fluss +connector's client compiles its protocol definitions in a build script. + +```bash +sudo apt-get install -y protobuf-compiler # Debian / Ubuntu +brew install protobuf # macOS +``` + +Alternatively, point `PROTOC` at an existing binary. + ### Quality Checks For Rust code: