ryerraguntla opened a new issue, #3688: URL: https://github.com/apache/iggy/issues/3688
### Description Add a **source connector** that reads from [Apache Fluss](https://fluss.apache.org/) tables and publishes records into Iggy streams/topics. Fluss is streaming storage for real-time analytics and lakehouse hot tiers. Its official Rust client ([`fluss-rs`](https://crates.io/crates/fluss-rs)) uses **Apache Arrow** (`RecordBatch` / `GenericRow`) end-to-end. That aligns with Iggy's connectors subsystem and with **Arrow-friendly** downstream sinks (for example Iceberg). Use cases: - Bridge Fluss log-table changelogs into Iggy for fan-out to multiple consumers or SDKs. - Ingest Fluss primary-key table CDC-style changes into Iggy for event-driven apps. - Unify Fluss hot-tier data with Iggy-based routing, transforms, and connector egress. ### Affected area / component Connectors ### Proposed solution Implement `iggy_connector_fluss_source` as a `cdylib` plugin implementing `iggy_connector_sdk::Source`, following conventions from `postgres_source` and `influxdb_source`. ### Client dependency Use [`fluss-rs`](https://github.com/apache/fluss-rust) (Tokio async, Arrow-native). Pin a released version in the workspace `Cargo.toml` when adding the crate. ### Table modes (phased) | Phase | Fluss table type | Read API | Notes | | ----- | ---------------- | -------- | ----- | | 1 | Log table | `subscribe` + `poll` / batch scanner | Offset-ordered append-only; primary use case | | 2 | Primary-key table | Changelog / scan APIs as exposed by `fluss-rs` | Upsert/delete events; confirm API stability in 0.1.x | ### Message mapping Default: decode Fluss rows to **`Schema::Json`** via existing connector JSON encoders (one object per row, column names from Fluss schema). Optional (config flag): emit **`Schema::Raw`** Arrow IPC bytes per `RecordBatch` for zero-copy analytics pipelines (consumers decode Arrow client-side). ### Offset / state Persist Fluss log offsets (and partition/bucket identifiers if applicable) in connector state via the runtime state API (`msgpack`), same pattern as other sources. On restart, resume from last committed offset. ### Draft configuration ```toml [plugin_config] bootstrap_servers = "localhost:9123" database = "mydb" table = "events" table_type = "log" # "log" | "primary_key" starting_offset = "earliest" # "earliest" | "latest" | explicit offset batch_size = 1000 poll_interval = "500ms" payload_format = "json" # "json" | "arrow_ipc" # Optional auth when supported by cluster # sasl_mechanism = "PLAIN" # sasl_username = "..." # sasl_password = "..." ``` Implement `iggy_connector_fluss_source` as a `cdylib` plugin implementing `iggy_connector_sdk::Source`, following conventions from `postgres_source` and `influxdb_source`. ### Client dependency Use [`fluss-rs`](https://github.com/apache/fluss-rust) (Tokio async, Arrow-native). Pin a released version in the workspace `Cargo.toml` when adding the crate. ### Table modes (phased) | Phase | Fluss table type | Read API | Notes | | ----- | ---------------- | -------- | ----- | | 1 | Log table | `subscribe` + `poll` / batch scanner | Offset-ordered append-only; primary use case | | 2 | Primary-key table | Changelog / scan APIs as exposed by `fluss-rs` | Upsert/delete events; confirm API stability in 0.1.x | ### Message mapping Default: decode Fluss rows to **`Schema::Json`** via existing connector JSON encoders (one object per row, column names from Fluss schema). Optional (config flag): emit **`Schema::Raw`** Arrow IPC bytes per `RecordBatch` for zero-copy analytics pipelines (consumers decode Arrow client-side). ### Offset / state Persist Fluss log offsets (and partition/bucket identifiers if applicable) in connector state via the runtime state API (`msgpack`), same pattern as other sources. On restart, resume from last committed offset. ### Draft configuration ```toml [plugin_config] bootstrap_servers = "localhost:9123" database = "mydb" table = "events" table_type = "log" # "log" | "primary_key" starting_offset = "earliest" # "earliest" | "latest" | explicit offset batch_size = 1000 poll_interval = "500ms" payload_format = "json" # "json" | "arrow_ipc" # Optional auth when supported by cluster # sasl_mechanism = "PLAIN" # sasl_username = "..." # sasl_password = "..." ``` ### Files / layout (expected) ```text core/connectors/sources/fluss_source/ ├── Cargo.toml ├── config.toml ├── README.md └── src/lib.rs ``` Register in workspace `Cargo.toml`, `core/connectors/sources/README.md`, and add a runtime example under `core/connectors/runtime/example_config/connectors/`. ### Example plugins - Structure / polling loop: `postgres_source` - External client + batching: `influxdb_source` - Arrow batch handling (sink side, for reference): `iceberg_sink` ## Acceptance criteria - [ ] `fluss_source` plugin builds as `cdylib` and loads in connectors runtime - [ ] Log-table scan reads batches and publishes to configured Iggy stream/topic - [ ] Offset checkpointing survives connector restart - [ ] Config documented in plugin `README.md` with minimal runnable example - [ ] Unit tests for config parsing and row-to-JSON mapping; integration test or documented manual test plan for live Fluss cluster - [ ] `cargo fmt`, `cargo sort --no-format`, `cargo clippy`, `cargo test -p iggy_connector_fluss_source` pass ## References - Apache Fluss: https://fluss.apache.org/ - Fluss Rust client: https://github.com/apache/fluss-rust - `fluss-rs` on crates.io: https://crates.io/crates/fluss-rs - Fluss + Arrow design: https://fluss.apache.org/blog/why-fluss-chose-rust-for-multi-language-sdk/ - Related: sink connector issue (file separately; link when available) ### Alternatives considered 1. **Flink-only integration** — Fluss already has a Flink connector; does not help users who standardize on Iggy connectors runtime. 2. **Generic HTTP/gRPC source** — Fluss is not HTTP-first; official Rust client is the supported path. 3. **Java client via JNI** — Iggy connectors are Rust `cdylib` plugins; `fluss-rs` matches the stack. ### Contribution - [ ] I'm willing to submit a pull request to implement this feature ### Good first issue - [ ] I think this could be a good first issue for a new contributor -- 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]
