add apache Fluss source connector

[ryerraguntla]

Description

Add a source connector that reads from Apache Fluss 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) 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 (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

[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 (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

[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)

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

[seokjin0414]

@hubcio
can i take this?

[seokjin0414]

@hubcio a build-system question before I start, since the answer changes the shape of the PR.

fluss-rs 0.1.0 (the only non-yanked release) compiles a bundled .proto in its build script, so it needs a system protoc. On a clean machine:

error: failed to run custom build command for `fluss-rs v0.1.0`
  --- stderr
  Error: Custom { kind: NotFound, error: "Could not find `protoc`. If `protoc` is
  installed, try setting the `PROTOC` environment variable to the path of the
  `protoc` binary. ..." }

prost-build 0.14 dropped its vendored feature and the crate ships the .proto rather than generated code, so there's no way around it from our side.

Nothing in .github/ or scripts/ installs protoc today, and ubuntu-latest (24.04) doesn't ship it either. The workspace has no default-members, so both cargo check --all --all-features and cargo clippy --all-features --all-targets select every member. Adding core/connectors/sources/fluss_source as a normal member therefore means CI and local dev both need protoc installed.

Two ways I see:

1. Add it as a normal member, install protobuf-compiler in CI, document it in CONTRIBUTING. One line in each, and the connector stays consistent with the others. The cost is that contributors need protoc for the standard cargo clippy --all-features --all-targets step.

2. Put it in exclude alongside foreign/cpp. The workspace build is untouched, but excluded packages can't inherit [workspace.dependencies] (that's why foreign/python hard-pins everything), so every dep gets pinned locally and the crate falls outside cargo sort --workspace, bump-version.sh and the cargo-rail scoping. A dedicated job would still need protoc.

Dependency-wise it's fairly contained: 28 new external packages in Cargo.lock, most of them prost-build's build-time deps, and fluss-rs's arrow ^57 resolves onto the 57.3.1 tree the workspace already carries alongside 58.3.0. I checked by adding the crate locally and diffing Cargo.lock.

I'd go with 1, mainly to keep the connector conventions intact, but adding a system build dependency to the repo seems like your call rather than mine. Whichever you pick also decides the integration-test shape, since testcontainers-modules has no Fluss module. A hand-rolled three-container fixture does work (I brought one up locally against 0.9.1-incubating), it's just more moving parts than the existing connector tests.

Scope note: 0.1.0 is from April and 0.2.0 was yanked on release day, so the pinned version sits a fair way behind main. It does work against a current server though: I ran it against a 0.9.1-incubating cluster and create table, append, scan, the Arrow batch scanner and column projection all round-trip fine. The first PR would cover log tables -> Schema::Json. arrow_ipc payloads are doable since the RecordBatch scanner is in 0.1.0, but it's a separate scanner with a separate offset-tracking path, so I'd rather leave that for a follow-up. PK changelog scanning is main-only, so phase 2 waits for a later fluss-rs release.

[hubcio]

thanks for digging into this. i don't like either option though, both assume we have to take on protoc as a build requirement, and i don't think we do.

one correction: prost-build didn't drop vendored in 0.14, that happened in 0.11 back in 2022 (the feature only ever existed in 0.10.x). today the vendored protoc trick comes from separate crates like protobuf-src or protoc-bin-vendored, which a build script has to use itself. you're still right that we can't fix it from our side, but it shows where the fix belongs.

the real problem is upstream. fluss_api.proto is one static file that never changes between builds, so every user of the crate is forced to run codegen for no reason. the normal fix is to commit the generated code into the repo: prost-types does this, and arrow-rs (already a dependency of fluss-rs) commits its generated flatbuffers code with a script to regenerate it. fluss-rs could do the same, or at least use protoc-bin-vendored in its build.rs as a fallback.

so let's fix it in apache/fluss-rust instead of working around it here. i don't want everyone who runs cargo clippy --all-features --all-targets to need protoc because of one optional plugin, and exclude creates a crate that misses out on workspace deps, cargo sort, version bumps and ci scoping. would you open an issue there (ideally with the pr, it looks small)? i'll back it on their side. you already showed 0.1.0 works against a 0.9.1 server, so a simple 0.1.1 patch release would unblock us, nothing needed from their main branch. once that ships, the connector comes in as a normal workspace member with zero ci changes. if upstream takes too long, we just wait.

on scope: agreed on everything, log tables into Schema::Json first, arrow_ipc later, pk changelog waits for a newer client. hand-rolled test setup is fine.

or, am I missing something? please correct me if im wrong

@seokjin0414

[seokjin0414]

You're right about the timeline, thanks for the correction. vendored only ever existed up to prost-build 0.10.x (0.10.4 still has the feature, 0.11.0 doesn't), so my "0.14 dropped it" was off by a couple of years.

Agreed on fixing this upstream, and it's done on my side: apache/fluss-rust#661 with the PR at apache/fluss-rust#662, following the arrow-flight pattern (checked-in prost output, a small gen crate, regen.sh for maintainers).

• the generated proto.rs (1,343 lines) is checked in next to fluss_api.proto, byte-identical to what their build.rs produces today
• build.rs and the prost-build build-dependency are gone; the gen crate depends on prost-build as a normal dependency, so building the workspace never invokes protoc, only running the regeneration does
• cargo build, clippy --all-targets --workspace -- -D warnings and the 572 unit tests pass on a machine with no protoc installed

Two things I found on the way: their own CI installs protoc across seven workflows (release_python.yml alone has five per-platform install blocks), all of which becomes deletable; and their main has had check_license_and_formatting red since Jul 15 because Rust 1.97 clippy flags two for_kv_map cases in accumulator.rs, so the PR fixes those first to be able to go green at all.

Once a 0.1.1 with the fix ships, I'll drop the two protoc commits from #3799 (the CI install and the CONTRIBUTING note) and rebase; the connector and tests themselves don't change.