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: 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 } 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)); + } +} 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;