From af17922c4c697cfdce15afec08226e9e98e8ab19 Mon Sep 17 00:00:00 2001 From: Adam Recsko Date: Thu, 30 Jul 2026 18:12:20 +0200 Subject: [PATCH] feat(connectors): add Apache Fluss sink Relates to #3689 --- Cargo.lock | 382 +++++++++++- Cargo.toml | 2 + core/connectors/sinks/fluss_sink/Cargo.toml | 46 ++ core/connectors/sinks/fluss_sink/README.md | 236 ++++++++ core/connectors/sinks/fluss_sink/config.toml | 60 ++ .../connectors/sinks/fluss_sink/src/config.rs | 230 ++++++++ core/connectors/sinks/fluss_sink/src/lib.rs | 147 +++++ .../connectors/sinks/fluss_sink/src/schema.rs | 545 ++++++++++++++++++ .../connectors/sinks/fluss_sink/src/writer.rs | 186 ++++++ core/integration/Cargo.toml | 1 + .../connectors/fixtures/fluss/cluster.rs | 228 ++++++++ .../tests/connectors/fixtures/fluss/mod.rs | 29 + .../tests/connectors/fixtures/fluss/sink.rs | 190 ++++++ .../tests/connectors/fixtures/mod.rs | 2 + .../tests/connectors/fluss/fluss_sink.rs | 239 ++++++++ .../integration/tests/connectors/fluss/mod.rs | 18 + .../tests/connectors/fluss/sink.toml | 20 + core/integration/tests/connectors/mod.rs | 2 +- 18 files changed, 2551 insertions(+), 12 deletions(-) create mode 100644 core/connectors/sinks/fluss_sink/Cargo.toml create mode 100644 core/connectors/sinks/fluss_sink/README.md create mode 100644 core/connectors/sinks/fluss_sink/config.toml create mode 100644 core/connectors/sinks/fluss_sink/src/config.rs create mode 100644 core/connectors/sinks/fluss_sink/src/lib.rs create mode 100644 core/connectors/sinks/fluss_sink/src/schema.rs create mode 100644 core/connectors/sinks/fluss_sink/src/writer.rs create mode 100644 core/integration/tests/connectors/fixtures/fluss/cluster.rs create mode 100644 core/integration/tests/connectors/fixtures/fluss/mod.rs create mode 100644 core/integration/tests/connectors/fixtures/fluss/sink.rs create mode 100644 core/integration/tests/connectors/fluss/fluss_sink.rs create mode 100644 core/integration/tests/connectors/fluss/mod.rs create mode 100644 core/integration/tests/connectors/fluss/sink.toml diff --git a/Cargo.lock b/Cargo.lock index 0ecf6332b7..5f772b49ab 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", @@ -3961,6 +4012,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" @@ -4022,7 +4084,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", @@ -4030,7 +4092,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", @@ -4873,7 +4935,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", @@ -4990,6 +5052,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" @@ -5038,12 +5106,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" @@ -5949,6 +6063,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" @@ -5957,7 +6080,7 @@ checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.2.0", ] [[package]] @@ -6960,6 +7083,21 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_fluss_sink" +version = "0.4.1-edge.1" +dependencies = [ + "async-trait", + "fluss-rs", + "iggy_common", + "iggy_connector_sdk", + "secrecy", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "iggy_connector_http_sink" version = "0.4.1-edge.1" @@ -7431,6 +7569,7 @@ dependencies = [ "deltalake", "dtor 1.0.5", "figment", + "fluss-rs", "futures", "harness_derive", "humantime", @@ -7570,10 +7709,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", ] @@ -8084,6 +8225,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" @@ -8637,6 +8781,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" @@ -9272,6 +9422,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" @@ -9439,7 +9600,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", ] @@ -9458,6 +9630,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" @@ -9650,6 +9836,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" @@ -10092,6 +10289,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" @@ -10373,6 +10589,7 @@ dependencies = [ "libc", "rand_chacha 0.3.1", "rand_core 0.6.4", + "serde", ] [[package]] @@ -10423,6 +10640,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ "getrandom 0.2.17", + "serde", ] [[package]] @@ -11603,6 +11821,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" @@ -12617,6 +12844,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" @@ -12635,6 +12868,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" @@ -12665,6 +12911,84 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "sval" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fb9efbae90f97301f4d25f3be63dfd99d6b7af9d088228a52ec960d649b2e7d" + +[[package]] +name = "sval_buffer" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff5c0280ea0af40b3a1fd0b532680b5067482ffb81412ee66ec04b1d9952b49a" +dependencies = [ + "sval", + "sval_ref", +] + +[[package]] +name = "sval_dynamic" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59b9067f2e68f58e110cf8019a268057e3791cf74b0fdb1b3c7c6e49104f44e6" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_fmt" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96ebdbf0e4b175884aa587fcf551c16dabe245ea04235abdd8cbbd160b27ed5a" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_json" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e448d9fa216a6c16670b28d624fbbcf5c04e41eb187bd7c52e01222ffa72a12" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_nested" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "021de5b5c26efd544c694cef9b8a9abe8633481bf3be1ea145a18a740818b291" +dependencies = [ + "sval", + "sval_buffer", + "sval_ref", +] + +[[package]] +name = "sval_ref" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54ef5ffec8bc52ded04ee424ab8d959e25e64fd40a48a16d21f8991ce824c1bb" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_serde" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b983833a8a2390f89ebcf9b9acd06883017014b4ffd72ee28e0c9de6852039f5" +dependencies = [ + "serde_core", + "sval", + "sval_nested", +] + [[package]] name = "svgtypes" version = "0.15.3" @@ -12968,7 +13292,7 @@ dependencies = [ "itertools 0.14.0", "log", "memchr", - "parse-display", + "parse-display 0.9.1", "pin-project-lite", "reqwest 0.13.4", "serde", @@ -14110,6 +14434,42 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "value-bag" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef73bfbaf3216cb59c205d7176bee1194e0d84348979da31f4a71fefe3c2054e" +dependencies = [ + "value-bag-serde1", + "value-bag-sval2", +] + +[[package]] +name = "value-bag-serde1" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b92170db3db8a6354f12a5b7f13a5453928433e08fc46aee51eedfa8f7a28a1" +dependencies = [ + "erased-serde", + "serde_core", + "serde_fmt", +] + +[[package]] +name = "value-bag-sval2" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf9832097ca044466ae3f1aa43a943d4e450b7c3b8cdf65363875df07da79a0" +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 0147c62ed6..28bf910490 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,6 +37,7 @@ members = [ "core/connectors/sinks/delta_sink", "core/connectors/sinks/doris_sink", "core/connectors/sinks/elasticsearch_sink", + "core/connectors/sinks/fluss_sink", "core/connectors/sinks/http_sink", "core/connectors/sinks/iceberg_sink", "core/connectors/sinks/influxdb_sink", @@ -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/sinks/fluss_sink/Cargo.toml b/core/connectors/sinks/fluss_sink/Cargo.toml new file mode 100644 index 0000000000..d4df9f9e0f --- /dev/null +++ b/core/connectors/sinks/fluss_sink/Cargo.toml @@ -0,0 +1,46 @@ +# 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_sink" +version = "0.4.1-edge.1" +description = "Iggy Apache Fluss sink connector for storing stream messages into Apache Fluss storage" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "fluss", "sink"] +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 + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +fluss-rs = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/core/connectors/sinks/fluss_sink/README.md b/core/connectors/sinks/fluss_sink/README.md new file mode 100644 index 0000000000..857ad7ddc4 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/README.md @@ -0,0 +1,236 @@ +# Apache Fluss Sink Connector + +The Apache Fluss sink connector consumes messages from Apache Iggy streams and +appends them to an [Apache Fluss](https://fluss.apache.org/) log table. It can +create the target table automatically, preserve selected Iggy metadata, and +store payloads as either Fluss `BYTES` or `STRING`. + +The connector uses the +[Fluss Rust client](https://clients.fluss.apache.org/user-guide/rust/api-reference/) +and exposes its writer, connection, and security configuration. + +## Features + +- Appends Iggy messages to a Fluss log table. +- Creates the target table on demand. +- Flushes all pending writes before each consumed batch completes. +- Supports Fluss writer retries, idempotence, buffering, and backpressure. +- Supports `PLAINTEXT` and SASL `PLAIN` client configuration. +- Optionally stores Iggy checksum, origin timestamp, and stream metadata. +- Stores payloads as Fluss `BYTES` or `STRING`. + +## Build + +From the repository root: + +```bash +cargo build --release -p iggy_connector_fluss_sink +``` + +The connector runtime loads the resulting dynamic library from +`target/release/`. Adjust the `path` setting for the working directory and +operating system used by the runtime. + +## Configuration + +The following example uses the default plugin settings explicitly. See +[`config.toml`](config.toml) for the complete configuration file. + +```toml +type = "sink" +key = "fluss" +enabled = true +version = 0 +name = "Fluss sink" +path = "target/release/libiggy_connector_fluss_sink" +verbose = false + +[[streams]] +stream = "user_events" +topics = ["users", "orders"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "fluss_sink" + +[plugin_config] +bootstrap_servers = "127.0.0.1:9123" +target_database = "fluss" +target_table = "iggy_messages" +auto_create_table = true +include_metadata = true +include_checksum = true +include_origin_timestamp = true +payload_format = "json" +``` + +All plugin fields have defaults. Existing configurations remain valid when new +fields are added because missing fields use the connector defaults. + +### Connector settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `target_database` | string | `"fluss"` | Target Fluss database. The database must already exist. | +| `target_table` | string | `"iggy_messages"` | Target Fluss table. | +| `auto_create_table` | bool | `true` | Create the target table if it does not exist before writing a batch. Existing tables are left unchanged. | +| `include_metadata` | bool | `true` | Add the Iggy offset, timestamp, stream, topic, and partition columns. | +| `include_checksum` | bool | `true` | Add the Iggy message checksum column. | +| `include_origin_timestamp` | bool | `true` | Add the Iggy origin timestamp column. | +| `payload_format` | enum | `"json"` | Payload storage format: `bytea`, `json`, or `text`. | + +### Fluss writer and connection settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `bootstrap_servers` | string | `"127.0.0.1:9123"` | Fluss coordinator address. | +| `writer_request_max_size` | i32 | `10485760` | Maximum writer request size in bytes. | +| `writer_acks` | string | `"all"` | Required acknowledgements. `"all"` waits for all required replicas. | +| `writer_retries` | i32 | `2147483647` | Maximum retries for transient writer failures. | +| `writer_batch_size` | i32 | `2097152` | Target Fluss writer batch size in bytes. | +| `writer_batch_timeout_ms` | i64 | `100` | Maximum time to wait for a writer batch to fill before sending it. | +| `writer_bucket_no_key_assigner` | enum | `"sticky"` | Bucket selection for tables without bucket keys: `sticky` or `round_robin`. | +| `writer_enable_idempotence` | bool | `true` | Add writer IDs and per-bucket sequence numbers so Fluss can deduplicate retried batches. | +| `writer_max_inflight_requests_per_bucket` | usize | `5` | Maximum unacknowledged requests per bucket. Idempotent writes require a value no greater than `5`. | +| `writer_buffer_memory_size` | usize | `67108864` | Total memory in bytes available for buffered write batches. | +| `writer_buffer_wait_timeout_ms` | string | `"18446744073709551615"` | Maximum time to wait for writer buffer memory. This is a string because the default is `u64::MAX`, which TOML integers cannot represent. | +| `connect_timeout_ms` | u64 | `120000` | TCP connection timeout in milliseconds. | + +Idempotent writes require `writer_acks = "all"` or `"-1"`, +`writer_retries > 0`, and +`writer_max_inflight_requests_per_bucket <= 5`. + +### Security settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `security_protocol` | string | `"PLAINTEXT"` | Use `"PLAINTEXT"` without authentication or `"sasl"` for SASL authentication. Matching is case-insensitive. | +| `security_sasl_mechanism` | string | `"PLAIN"` | SASL mechanism. The pinned Fluss client supports only `PLAIN`. | +| `security_sasl_username` | string | `""` | SASL username. Required when `security_protocol = "sasl"`. | +| `security_sasl_password` | string | `""` | SASL password. Required when `security_protocol = "sasl"` and omitted from serialized connector configuration. | + +Example: + +```toml +[plugin_config] +bootstrap_servers = "fluss.example.com:9123" +security_protocol = "sasl" +security_sasl_mechanism = "PLAIN" +security_sasl_username = "iggy" +security_sasl_password = "replace-with-secret" +``` + +## Payload formats + +| Value | Fluss type | Behavior | +| --- | --- | --- | +| `bytea` | `BYTES` | Serializes the Iggy payload to bytes and preserves it in a binary column. | +| `json` | `STRING` | Serializes the payload to bytes and stores the resulting UTF-8 string. | +| `text` | `STRING` | Serializes the payload to bytes and stores the resulting UTF-8 string. | + +`json` and `text` currently use the same Fluss schema and row conversion. The +sink does not parse or validate JSON itself. Configure the Iggy stream with +`schema = "json"` when JSON validation is required before the sink receives the +message. + +Any Iggy payload variant can be written with `bytea`. The `json` and `text` +formats reject payload bytes that are not valid UTF-8. + +## Generated table schema + +When `auto_create_table = true`, the connector creates an append-only Fluss log +table without a primary key. Columns are generated in the following order: + +| Column | Fluss type | Included when | +| --- | --- | --- | +| `id` | `STRING` | Always | +| `checksum` | `DECIMAL(20, 0)` | `include_checksum = true` | +| `iggy_offset` | `DECIMAL(20, 0)` | `include_metadata = true` | +| `iggy_timestamp` | `TIMESTAMP_LTZ(6)` | `include_metadata = true` | +| `iggy_stream` | `STRING` | `include_metadata = true` | +| `iggy_topic` | `STRING` | `include_metadata = true` | +| `iggy_partition_id` | `BIGINT` | `include_metadata = true` | +| `iggy_origin_timestamp` | `TIMESTAMP_LTZ(6)` | `include_origin_timestamp = true` | +| `payload` | `BYTES` or `STRING` | Always | + +Message IDs are encoded as 32-character lowercase hexadecimal strings. +`DECIMAL(20, 0)` preserves the complete unsigned 64-bit range for offsets and +checksums. Timestamps are interpreted as microseconds since the Unix epoch. + +### Manual table creation with Flink SQL + +When `auto_create_table = false`, create the database and table through a +[Fluss catalog in Flink SQL](https://fluss.apache.org/docs/engine-flink/getting-started/) +before starting the connector. The following definition matches the default +`target_database = "fluss"`, `target_table = "iggy_messages"`, and +`payload_format = "json"` settings: + +```sql +USE CATALOG fluss_catalog; + +CREATE DATABASE IF NOT EXISTS `fluss`; +USE `fluss`; + +CREATE TABLE `iggy_messages` ( + `id` STRING COMMENT 'Apache Iggy message ID', + `checksum` DECIMAL(20, 0) COMMENT 'Apache Iggy message checksum', + `iggy_offset` DECIMAL(20, 0) COMMENT 'Apache Iggy message offset', + `iggy_timestamp` TIMESTAMP_LTZ(6) + COMMENT 'Apache Iggy message timestamp', + `iggy_stream` STRING COMMENT 'Apache Iggy stream name', + `iggy_topic` STRING COMMENT 'Apache Iggy topic name', + `iggy_partition_id` BIGINT COMMENT 'Apache Iggy partition ID', + `iggy_origin_timestamp` TIMESTAMP_LTZ(6) + COMMENT 'Apache Iggy message origin timestamp', + `payload` STRING COMMENT 'Apache Iggy message payload' +) +COMMENT 'Stores Apache Iggy messages written by the Fluss sink connector'; +``` + +Replace `fluss_catalog` with the name of the Fluss catalog configured in the +Flink SQL client. If `payload_format = "bytea"`, define `payload` as `BYTES` +instead of `STRING`. + +Omit `checksum` when `include_checksum = false`. Omit `iggy_offset`, +`iggy_timestamp`, `iggy_stream`, `iggy_topic`, and `iggy_partition_id` when +`include_metadata = false`. Omit `iggy_origin_timestamp` when +`include_origin_timestamp = false`. Keep the remaining columns in the order +shown above. + +Message headers are not stored. + +The connector does not migrate or alter existing tables. A manually created +table must use the same column order and compatible Fluss data types. +Tables created by connector versions that used `STRING` for checksum, offset, +and timestamps must be recreated or migrated before using this schema. + +## Write behavior + +For every batch received from the Iggy connector runtime, the sink: + +1. Creates the table if `auto_create_table` is enabled and the table is missing. +2. Opens an append writer for the target table. +3. Converts each Iggy message to a Fluss row. +4. Appends every row and flushes the writer. + +The effective message count per call is controlled by the stream +`batch_length`. Fluss may combine those rows into byte-sized writer batches +according to `writer_batch_size` and `writer_batch_timeout_ms`. + +## Limitations + +- The connector writes append-only log tables and does not support primary-key + upserts. +- The target database is not created automatically. +- Existing table schemas are not migrated. +- The pinned `fluss-rs` 0.1 client does not expose a public graceful connection + shutdown method. Each consumed batch is flushed before returning, and + connector shutdown currently releases the client by dropping it. + +## Testing + +Run the Fluss sink unit tests from the repository root: + +```bash +cargo test -p iggy_connector_fluss_sink +``` diff --git a/core/connectors/sinks/fluss_sink/config.toml b/core/connectors/sinks/fluss_sink/config.toml new file mode 100644 index 0000000000..404e8d87a3 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/config.toml @@ -0,0 +1,60 @@ +# 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 = "sink" +key = "fluss" +enabled = true +version = 0 +name = "Fluss sink" +path = "../../target/release/libiggy_connector_fluss_sink" +verbose = false + +[[streams]] +stream = "user_events" +topics = ["users", "orders"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "fluss_sink" + +[plugin_config] +# iggy's fluss table settings +target_database = "fluss" +target_table = "iggy_messages" +auto_create_table = true +include_metadata = true +include_checksum = true +include_origin_timestamp = true +payload_format = "json" + +# fluss connection settings, passed to the fluss connector +bootstrap_servers = "127.0.0.1:9123" +writer_request_max_size = 10485760 +writer_acks = "all" +writer_retries = 2147483647 +writer_batch_size = 2097152 +writer_bucket_no_key_assigner = "sticky" +writer_batch_timeout_ms = 100 +writer_enable_idempotence = true +writer_max_inflight_requests_per_bucket = 5 +writer_buffer_memory_size = 67108864 +writer_buffer_wait_timeout_ms = "18446744073709551615" +connect_timeout_ms = 120000 +security_protocol = "PLAINTEXT" +security_sasl_mechanism = "PLAIN" +security_sasl_username = "" +security_sasl_password = "" diff --git a/core/connectors/sinks/fluss_sink/src/config.rs b/core/connectors/sinks/fluss_sink/src/config.rs new file mode 100644 index 0000000000..eaeac8c39b --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/config.rs @@ -0,0 +1,230 @@ +// 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 fluss::config::{Config as FlussConfig, NoKeyAssigner}; +use iggy_connector_sdk::Error; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PayloadFormat { + Bytea, + #[default] + Json, + Text, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +#[serde(default)] +pub struct FlussSinkConfig { + pub bootstrap_servers: String, + pub writer_request_max_size: i32, + pub writer_acks: String, + pub writer_retries: i32, + pub writer_batch_size: i32, + pub writer_bucket_no_key_assigner: NoKeyAssigner, + pub writer_batch_timeout_ms: i64, + pub writer_enable_idempotence: bool, + pub writer_max_inflight_requests_per_bucket: usize, + pub writer_buffer_memory_size: usize, + pub writer_buffer_wait_timeout_ms: String, + pub connect_timeout_ms: u64, + pub security_protocol: String, + pub security_sasl_mechanism: String, + pub security_sasl_username: String, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")] + pub security_sasl_password: SecretString, + pub target_database: String, + pub target_table: String, + pub auto_create_table: bool, + pub include_metadata: bool, + pub include_checksum: bool, + pub include_origin_timestamp: bool, + pub payload_format: PayloadFormat, +} + +impl Default for FlussSinkConfig { + fn default() -> Self { + let fluss_config = FlussConfig::default(); + Self { + bootstrap_servers: fluss_config.bootstrap_servers, + writer_request_max_size: fluss_config.writer_request_max_size, + writer_acks: fluss_config.writer_acks, + writer_retries: fluss_config.writer_retries, + writer_batch_size: fluss_config.writer_batch_size, + writer_bucket_no_key_assigner: fluss_config.writer_bucket_no_key_assigner, + writer_batch_timeout_ms: fluss_config.writer_batch_timeout_ms, + writer_enable_idempotence: fluss_config.writer_enable_idempotence, + writer_max_inflight_requests_per_bucket: fluss_config + .writer_max_inflight_requests_per_bucket, + writer_buffer_memory_size: fluss_config.writer_buffer_memory_size, + writer_buffer_wait_timeout_ms: fluss_config.writer_buffer_wait_timeout_ms.to_string(), + connect_timeout_ms: fluss_config.connect_timeout_ms, + security_protocol: fluss_config.security_protocol, + security_sasl_mechanism: fluss_config.security_sasl_mechanism, + security_sasl_username: fluss_config.security_sasl_username, + security_sasl_password: fluss_config.security_sasl_password.into(), + target_database: "fluss".to_string(), + target_table: "iggy_messages".to_string(), + auto_create_table: true, + include_metadata: true, + include_checksum: true, + include_origin_timestamp: true, + payload_format: PayloadFormat::default(), + } + } +} + +impl TryFrom<&FlussSinkConfig> for FlussConfig { + type Error = Error; + + fn try_from(config: &FlussSinkConfig) -> Result { + let writer_buffer_wait_timeout_ms = + config + .writer_buffer_wait_timeout_ms + .parse() + .map_err(|error| { + Error::InvalidConfigValue(format!( + "invalid writer_buffer_wait_timeout_ms '{}': {error}", + config.writer_buffer_wait_timeout_ms + )) + })?; + + Ok(Self { + bootstrap_servers: config.bootstrap_servers.clone(), + writer_request_max_size: config.writer_request_max_size, + writer_acks: config.writer_acks.clone(), + writer_retries: config.writer_retries, + writer_batch_size: config.writer_batch_size, + writer_bucket_no_key_assigner: config.writer_bucket_no_key_assigner, + writer_batch_timeout_ms: config.writer_batch_timeout_ms, + writer_enable_idempotence: config.writer_enable_idempotence, + writer_max_inflight_requests_per_bucket: config.writer_max_inflight_requests_per_bucket, + writer_buffer_memory_size: config.writer_buffer_memory_size, + writer_buffer_wait_timeout_ms, + connect_timeout_ms: config.connect_timeout_ms, + security_protocol: config.security_protocol.clone(), + security_sasl_mechanism: config.security_sasl_mechanism.clone(), + security_sasl_username: config.security_sasl_username.clone(), + security_sasl_password: config.security_sasl_password.expose_secret().to_string(), + ..FlussConfig::default() + }) + } +} + +#[cfg(test)] +mod tests { + use fluss::config::Config as FlussConfig; + use iggy_connector_sdk::Error; + use serde_json::json; + + use super::{FlussSinkConfig, PayloadFormat}; + + #[test] + fn given_default_sink_config_when_converting_should_match_fluss_defaults() { + let sink_config = FlussSinkConfig::default(); + let fluss_config = FlussConfig::try_from(&sink_config).expect("Sink config should convert"); + let actual = serde_json::to_value(fluss_config).expect("Fluss config should serialize"); + let expected = + serde_json::to_value(FlussConfig::default()).expect("Fluss config should serialize"); + + assert_eq!(actual, expected); + } + + #[test] + fn given_existing_sink_config_when_deserializing_should_apply_fluss_defaults() { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "bootstrap_servers": "localhost:9123", + "target_database": "analytics", + "target_table": "events", + "auto_create_table": true, + "include_metadata": true, + "include_checksum": true, + "include_origin_timestamp": true, + "payload_format": "json" + })) + .expect("Existing Fluss sink config should deserialize"); + + assert_eq!(config.writer_batch_size, 2 * 1024 * 1024); + assert_eq!(config.writer_buffer_wait_timeout_ms, u64::MAX.to_string()); + assert_eq!(config.payload_format, PayloadFormat::Json); + assert_eq!(config.target_database, "analytics"); + assert_eq!(config.target_table, "events"); + } + + #[test] + fn given_supported_payload_formats_when_deserializing_should_return_matching_variants() { + for (value, expected) in [ + ("bytea", PayloadFormat::Bytea), + ("json", PayloadFormat::Json), + ("text", PayloadFormat::Text), + ] { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "payload_format": value + })) + .expect("Supported payload format should deserialize"); + + assert_eq!(config.payload_format, expected); + } + } + + #[test] + fn given_unsupported_payload_format_when_deserializing_should_fail() { + let config = serde_json::from_value::(json!({ + "payload_format": "xml" + })); + + assert!(config.is_err()); + } + + #[test] + fn given_u64_max_as_string_when_converting_should_parse_value() { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "writer_buffer_wait_timeout_ms": u64::MAX.to_string() + })) + .expect("String-encoded u64 should deserialize"); + let fluss_config = FlussConfig::try_from(&config).expect("Sink config should convert"); + + assert_eq!(fluss_config.writer_buffer_wait_timeout_ms, u64::MAX); + } + + #[test] + fn given_numeric_buffer_wait_timeout_when_deserializing_should_fail() { + let config = serde_json::from_value::(json!({ + "writer_buffer_wait_timeout_ms": 100 + })); + + assert!(config.is_err()); + } + + #[test] + fn given_invalid_buffer_wait_timeout_when_converting_should_return_invalid_config_value() { + let config = FlussSinkConfig { + writer_buffer_wait_timeout_ms: "invalid".to_string(), + ..FlussSinkConfig::default() + }; + + let error = FlussConfig::try_from(&config).expect_err("Invalid value should fail"); + + assert!(matches!( + error, + Error::InvalidConfigValue(message) + if message.contains("invalid writer_buffer_wait_timeout_ms 'invalid'") + )); + } +} diff --git a/core/connectors/sinks/fluss_sink/src/lib.rs b/core/connectors/sinks/fluss_sink/src/lib.rs new file mode 100644 index 0000000000..4d5c698ef0 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/lib.rs @@ -0,0 +1,147 @@ +// 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 async_trait::async_trait; +use fluss::metadata::TablePath; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use tokio::sync::Mutex; +use tracing::{debug, info}; + +use crate::{schema::FlussTableLayout, writer::FlussWriter}; + +mod config; +mod schema; +mod writer; + +pub use config::{FlussSinkConfig, PayloadFormat}; + +sink_connector!(FlussSink); + +#[derive(Debug)] +struct State { + invocations_count: u64, + insertion_errors: u64, + messages_processed: u64, +} + +#[derive(Debug)] +pub struct FlussSink { + id: u32, + state: Mutex, + fluss_writer: writer::FlussWriter, + fluss_config: FlussSinkConfig, + table_layout: Option, + table_path: TablePath, +} + +impl FlussSink { + pub fn new(id: u32, config: FlussSinkConfig) -> Self { + let table_path = + TablePath::new(config.target_database.clone(), config.target_table.clone()); + Self { + id, + state: Mutex::new(State { + invocations_count: 0, + messages_processed: 0, + insertion_errors: 0, + }), + fluss_writer: FlussWriter::new(config.clone()), + fluss_config: config, + table_layout: None, + table_path, + } + } +} + +#[async_trait] +impl Sink for FlussSink { + async fn open(&mut self) -> Result<(), Error> { + let table_layout = FlussTableLayout::from_config(&self.fluss_config)?; + self.fluss_writer.connect().await?; + + self.fluss_writer + .ensure_table_exists(&self.table_path, &table_layout) + .await?; + + self.table_layout = Some(table_layout); + info!("Opened Fluss sink connector ID: {}", self.id); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec, + ) -> Result<(), Error> { + let invocation = { + let mut state = self.state.lock().await; + state.invocations_count += 1; + state.invocations_count + }; + + debug!( + "Fluss sink connector ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition_id: {}, current_offset: {}, invocation: {}", + self.id, + messages.len(), + messages_metadata.schema, + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset, + invocation + ); + + let table_layout = self + .table_layout + .as_ref() + .ok_or_else(|| Error::InitError("Fluss table layout is not initialized".to_string()))?; + + match self + .fluss_writer + .write_to_table( + &self.table_path, + messages_metadata, + messages, + topic_metadata, + table_layout, + ) + .await + { + Ok(result) => { + let mut state = self.state.lock().await; + state.insertion_errors += result.insertion_errors; + state.messages_processed += result.messages_processed; + Ok(()) + } + + Err(e) => Err(e), + } + } + + async fn close(&mut self) -> Result<(), Error> { + // fluss-rs 0.1.0 exposes no public connection close API + let state = self.state.lock().await; + info!( + "Fluss sink ID: {} processed {} messages with {} errors", + self.id, state.messages_processed, state.insertion_errors + ); + Ok(()) + } +} diff --git a/core/connectors/sinks/fluss_sink/src/schema.rs b/core/connectors/sinks/fluss_sink/src/schema.rs new file mode 100644 index 0000000000..5d1ec11043 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/schema.rs @@ -0,0 +1,545 @@ +// 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 std::mem::size_of; + +use fluss::{ + error::Error as FlussError, + metadata::{Column, DataTypes, Schema, TableDescriptor}, + row::{Datum, Decimal, GenericRow, TimestampLtz}, +}; +use iggy_connector_sdk::{ConsumedMessage, Error}; + +use crate::{FlussSinkConfig, PayloadFormat}; + +const UNSIGNED_64_DECIMAL_PRECISION: u32 = 20; +const TIMESTAMP_PRECISION: u32 = 6; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ColumnKind { + MessageId, + Checksum, + MessageTimestamp, + OriginTimestamp, + MessageOffset, + Stream, + Topic, + PartitionId, + BinaryPayload, + StringPayload, +} + +impl From for Column { + fn from(column: ColumnKind) -> Self { + match column { + ColumnKind::MessageId => { + Column::new("id", DataTypes::string()).with_comment("Apache Iggy message ID") + } + ColumnKind::Checksum => Column::new( + "checksum", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION, 0), + ) + .with_comment("Apache Iggy message checksum"), + ColumnKind::MessageTimestamp => Column::new( + "iggy_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message timestamp"), + ColumnKind::OriginTimestamp => Column::new( + "iggy_origin_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message origin timestamp"), + ColumnKind::MessageOffset => Column::new( + "iggy_offset", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION, 0), + ) + .with_comment("Apache Iggy message offset"), + ColumnKind::Stream => Column::new("iggy_stream", DataTypes::string()) + .with_comment("Apache Iggy stream name"), + ColumnKind::Topic => Column::new("iggy_topic", DataTypes::string()) + .with_comment("Apache Iggy topic name"), + ColumnKind::PartitionId => Column::new("iggy_partition_id", DataTypes::bigint()) + .with_comment("Apache Iggy partition ID"), + ColumnKind::BinaryPayload => Column::new("payload", DataTypes::bytes()) + .with_comment("Apache Iggy message payload"), + ColumnKind::StringPayload => Column::new("payload", DataTypes::string()) + .with_comment("Apache Iggy message payload"), + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct RowContext<'a> { + pub stream: &'a str, + pub topic: &'a str, + pub partition_id: u32, +} + +impl ColumnKind { + fn datum<'a>( + self, + message: &'a ConsumedMessage, + context: RowContext<'a>, + ) -> Result, Error> { + match self { + Self::MessageId => Ok(format!("{:032x}", message.id).into()), + Self::Checksum => decimal_from_u64(message.checksum, "checksum").map(Into::into), + Self::MessageTimestamp => { + timestamp_from_micros(message.timestamp, "timestamp").map(Into::into) + } + Self::OriginTimestamp => { + timestamp_from_micros(message.origin_timestamp, "origin_timestamp").map(Into::into) + } + Self::MessageOffset => decimal_from_u64(message.offset, "offset").map(Into::into), + Self::Stream => Ok(context.stream.into()), + Self::Topic => Ok(context.topic.into()), + Self::PartitionId => Ok(i64::from(context.partition_id).into()), + Self::BinaryPayload => { + message + .payload + .try_to_bytes() + .map(Into::into) + .map_err(|error| { + Error::Serialization(format!( + "Failed to serialize payload for Fluss BYTES column: {error}" + )) + }) + } + Self::StringPayload => { + let payload_bytes = message.payload.try_to_bytes().map_err(|error| { + Error::Serialization(format!( + "Failed to serialize payload for Fluss STRING column: {error}" + )) + })?; + let payload_string = String::from_utf8(payload_bytes).map_err(|error| { + Error::Serialization(format!( + "Payload is not valid UTF-8 for the Fluss STRING column: {error}" + )) + })?; + Ok(payload_string.into()) + } + } + } +} + +fn decimal_from_u64(value: u64, field: &str) -> Result { + // Fluss reads signed two's-complement bytes, so the zero prefix preserves the high bit. + let mut unscaled_bytes = [0_u8; size_of::() + 1]; + unscaled_bytes[1..].copy_from_slice(&value.to_be_bytes()); + + Decimal::from_unscaled_bytes(&unscaled_bytes, UNSIGNED_64_DECIMAL_PRECISION, 0).map_err( + |error| { + Error::InvalidRecordValue(format!( + "Failed to convert Iggy {field} value {value} to Fluss DECIMAL(20, 0): {error}" + )) + }, + ) +} + +fn timestamp_from_micros(value: u64, field: &str) -> Result { + let epoch_micros = i64::try_from(value).map_err(|_| { + Error::InvalidRecordValue(format!( + "Iggy {field} value {value} exceeds the Fluss TIMESTAMP_LTZ(6) range" + )) + })?; + let epoch_millis = epoch_micros / 1_000; + let nanos_of_millisecond = ((epoch_micros % 1_000) * 1_000) as i32; + + TimestampLtz::from_millis_nanos(epoch_millis, nanos_of_millisecond).map_err(|error| { + Error::InvalidRecordValue(format!( + "Failed to convert Iggy {field} value {value} to Fluss TIMESTAMP_LTZ(6): {error}" + )) + }) +} + +#[derive(Debug)] +pub struct FlussTableLayout { + columns: Vec, + primary_key_columns: Vec, +} + +impl FlussTableLayout { + pub fn from_config(config: &FlussSinkConfig) -> Result { + let mut columns: Vec = Vec::with_capacity(10); + columns.push(ColumnKind::MessageId); + + if config.include_checksum { + columns.push(ColumnKind::Checksum); + }; + + if config.include_metadata { + columns.extend([ + ColumnKind::MessageOffset, + ColumnKind::MessageTimestamp, + ColumnKind::Stream, + ColumnKind::Topic, + ColumnKind::PartitionId, + ]); + }; + + if config.include_origin_timestamp { + columns.push(ColumnKind::OriginTimestamp); + } + + match config.payload_format { + PayloadFormat::Bytea => columns.push(ColumnKind::BinaryPayload), + PayloadFormat::Json | PayloadFormat::Text => { + columns.push(ColumnKind::StringPayload); + } + } + + Ok(Self { + columns, + primary_key_columns: Vec::new(), + }) + } + + fn build_schema(&self) -> Result { + let columns: Vec = self.columns.iter().copied().map(Into::into).collect(); + let mut schema_builder = Schema::builder().with_columns(columns); + + if !self.primary_key_columns.is_empty() { + schema_builder = schema_builder.primary_key(self.primary_key_columns.clone()); + } + + schema_builder.build() + } + + pub fn build_table_descriptor(&self) -> Result { + let schema = self.build_schema()?; + + TableDescriptor::builder() + .comment("Stores Apache Iggy messages written by the Fluss sink connector") + .schema(schema) + .build() + } + + pub fn row_from_message<'a>( + &self, + message: &'a ConsumedMessage, + context: RowContext<'a>, + ) -> Result, Error> { + let mut values: Vec = Vec::with_capacity(self.columns.len()); + for column in &self.columns { + values.push(column.datum(message, context)?); + } + Ok(GenericRow::from_data(values)) + } +} + +#[cfg(test)] +mod tests { + use fluss::{ + metadata::{Column, DataTypes}, + row::{Datum, TimestampLtz}, + }; + use iggy_connector_sdk::{ConsumedMessage, Error, Payload, Schema}; + + use super::{ + ColumnKind, FlussTableLayout, RowContext, TIMESTAMP_PRECISION, + UNSIGNED_64_DECIMAL_PRECISION, decimal_from_u64, + }; + use crate::{FlussSinkConfig, PayloadFormat}; + + const MESSAGE_TIMESTAMP: u64 = 1_700_000_000_123_456; + const ORIGIN_TIMESTAMP: u64 = 1_700_000_000_120_789; + + fn test_config(payload_format: PayloadFormat) -> FlussSinkConfig { + FlussSinkConfig { + payload_format, + ..FlussSinkConfig::default() + } + } + + fn config_without_optional_columns(payload_format: PayloadFormat) -> FlussSinkConfig { + FlussSinkConfig { + include_checksum: false, + include_metadata: false, + include_origin_timestamp: false, + ..test_config(payload_format) + } + } + + fn test_message(payload: Payload) -> ConsumedMessage { + ConsumedMessage { + id: 101, + offset: 202, + checksum: 303, + timestamp: MESSAGE_TIMESTAMP, + origin_timestamp: ORIGIN_TIMESTAMP, + headers: None, + payload, + } + } + + fn test_context() -> RowContext<'static> { + RowContext { + stream: "orders", + topic: "created", + partition_id: 7, + } + } + + #[test] + fn given_default_config_when_building_layout_should_include_all_columns_in_order() { + let layout = FlussTableLayout::from_config(&FlussSinkConfig::default()) + .expect("Default config should build a table layout"); + + assert_eq!( + layout.columns, + [ + ColumnKind::MessageId, + ColumnKind::Checksum, + ColumnKind::MessageOffset, + ColumnKind::MessageTimestamp, + ColumnKind::Stream, + ColumnKind::Topic, + ColumnKind::PartitionId, + ColumnKind::OriginTimestamp, + ColumnKind::StringPayload, + ] + ); + } + + #[test] + fn given_optional_columns_disabled_when_building_layout_should_only_include_id_and_payload() { + let config = config_without_optional_columns(PayloadFormat::Bytea); + let layout = FlussTableLayout::from_config(&config) + .expect("Config without optional columns should build a table layout"); + + assert_eq!( + layout.columns, + [ColumnKind::MessageId, ColumnKind::BinaryPayload] + ); + } + + #[test] + fn given_payload_formats_when_building_schema_should_use_matching_payload_types() { + for (payload_format, expected_data_type) in [ + (PayloadFormat::Bytea, DataTypes::bytes()), + (PayloadFormat::Json, DataTypes::string()), + (PayloadFormat::Text, DataTypes::string()), + ] { + let config = config_without_optional_columns(payload_format); + let layout = FlussTableLayout::from_config(&config) + .expect("Payload format should build a table layout"); + let schema = layout.build_schema().expect("Schema should build"); + let payload_column = schema + .columns() + .last() + .expect("Schema should contain a payload column"); + + assert_eq!(payload_column.name(), "payload"); + assert_eq!(payload_column.data_type(), &expected_data_type); + } + } + + #[test] + fn given_default_layout_when_building_descriptor_should_include_schema_metadata() { + let layout = FlussTableLayout::from_config(&FlussSinkConfig::default()) + .expect("Default config should build a table layout"); + let descriptor = layout + .build_table_descriptor() + .expect("Table descriptor should build"); + + assert_eq!( + descriptor.schema().columns(), + [ + Column::new("id", DataTypes::string()).with_comment("Apache Iggy message ID"), + Column::new( + "checksum", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION, 0), + ) + .with_comment("Apache Iggy message checksum"), + Column::new( + "iggy_offset", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION, 0), + ) + .with_comment("Apache Iggy message offset"), + Column::new( + "iggy_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message timestamp"), + Column::new("iggy_stream", DataTypes::string()) + .with_comment("Apache Iggy stream name"), + Column::new("iggy_topic", DataTypes::string()) + .with_comment("Apache Iggy topic name"), + Column::new("iggy_partition_id", DataTypes::bigint()) + .with_comment("Apache Iggy partition ID"), + Column::new( + "iggy_origin_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message origin timestamp"), + Column::new("payload", DataTypes::string()) + .with_comment("Apache Iggy message payload"), + ] + ); + assert_eq!( + descriptor.comment(), + Some("Stores Apache Iggy messages written by the Fluss sink connector") + ); + assert!(!descriptor.has_primary_key()); + } + + #[test] + fn given_primary_key_columns_when_building_schema_should_set_primary_key() { + let layout = FlussTableLayout { + columns: vec![ColumnKind::MessageId, ColumnKind::BinaryPayload], + primary_key_columns: vec!["id".to_string()], + }; + + let schema = layout.build_schema().expect("Schema should build"); + + assert_eq!(schema.primary_key_column_names(), ["id"]); + } + + #[test] + fn given_binary_payload_when_building_row_should_preserve_column_order_and_values() { + let layout = FlussTableLayout::from_config(&test_config(PayloadFormat::Bytea)) + .expect("Bytea config should build a table layout"); + let message = test_message(Payload::Raw(vec![0, 127, 255])); + + let row = layout + .row_from_message(&message, test_context()) + .expect("Binary row should build"); + + assert_eq!( + row.values, + [ + Datum::from("00000000000000000000000000000065".to_string()), + Datum::from( + decimal_from_u64(303, "checksum").expect("Checksum should convert to decimal") + ), + Datum::from( + decimal_from_u64(202, "offset").expect("Offset should convert to decimal") + ), + Datum::from( + TimestampLtz::from_millis_nanos(1_700_000_000_123, 456_000) + .expect("Timestamp should build") + ), + Datum::from("orders"), + Datum::from("created"), + Datum::from(7_i64), + Datum::from( + TimestampLtz::from_millis_nanos(1_700_000_000_120, 789_000) + .expect("Origin timestamp should build") + ), + Datum::from(vec![0, 127, 255]), + ] + ); + } + + #[test] + fn given_text_payload_when_building_row_should_store_payload_as_string() { + let config = config_without_optional_columns(PayloadFormat::Text); + let layout = FlussTableLayout::from_config(&config) + .expect("Text config should build a table layout"); + let message = test_message(Payload::Text("hello Fluss".to_string())); + + let row = layout + .row_from_message(&message, test_context()) + .expect("Text row should build"); + + assert_eq!( + row.values, + [ + Datum::from("00000000000000000000000000000065".to_string()), + Datum::from("hello Fluss".to_string()), + ] + ); + } + + #[test] + fn given_json_payload_when_building_row_should_serialize_payload_as_string() { + let config = config_without_optional_columns(PayloadFormat::Json); + let layout = FlussTableLayout::from_config(&config) + .expect("JSON config should build a table layout"); + let payload = Schema::Json + .try_into_payload(br#"{"event":"created"}"#.to_vec()) + .expect("JSON payload should decode"); + let message = test_message(payload); + + let row = layout + .row_from_message(&message, test_context()) + .expect("JSON row should build"); + + assert_eq!( + row.values, + [ + Datum::from("00000000000000000000000000000065".to_string()), + Datum::from(r#"{"event":"created"}"#.to_string()), + ] + ); + } + + #[test] + fn given_max_unsigned_values_when_building_row_should_preserve_id_offset_and_checksum() { + let layout = FlussTableLayout::from_config(&test_config(PayloadFormat::Bytea)) + .expect("Bytea config should build a table layout"); + let mut message = test_message(Payload::Raw(vec![1])); + message.id = u128::MAX; + message.offset = u64::MAX; + message.checksum = u64::MAX; + + let row = layout + .row_from_message(&message, test_context()) + .expect("Unsigned values should build"); + + assert_eq!(row.values[0].as_str(), "ffffffffffffffffffffffffffffffff"); + assert_eq!(row.values[1].as_decimal().to_string(), u64::MAX.to_string()); + assert_eq!(row.values[2].as_decimal().to_string(), u64::MAX.to_string()); + } + + #[test] + fn given_timestamp_above_fluss_range_when_building_row_should_return_invalid_record_value() { + let layout = FlussTableLayout::from_config(&test_config(PayloadFormat::Bytea)) + .expect("Bytea config should build a table layout"); + let mut message = test_message(Payload::Raw(vec![1])); + message.timestamp = i64::MAX as u64 + 1; + + let error = layout + .row_from_message(&message, test_context()) + .expect_err("Out-of-range timestamp should fail"); + + assert!(matches!( + error, + Error::InvalidRecordValue(message) + if message.contains("timestamp") && message.contains("TIMESTAMP_LTZ(6) range") + )); + } + + #[test] + fn given_invalid_utf8_when_building_string_row_should_return_serialization_error() { + let config = config_without_optional_columns(PayloadFormat::Text); + let layout = FlussTableLayout::from_config(&config) + .expect("Text config should build a table layout"); + let message = test_message(Payload::Raw(vec![0xFF])); + + let error = layout + .row_from_message(&message, test_context()) + .expect_err("Invalid UTF-8 payload should fail"); + + assert!(matches!( + error, + Error::Serialization(message) + if message.contains("not valid UTF-8 for the Fluss STRING column") + )); + } +} diff --git a/core/connectors/sinks/fluss_sink/src/writer.rs b/core/connectors/sinks/fluss_sink/src/writer.rs new file mode 100644 index 0000000000..4f566d61d3 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/writer.rs @@ -0,0 +1,186 @@ +// 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 std::fmt::{self, Display, Formatter}; + +use fluss::{ + client::FlussConnection, + metadata::{TableDescriptor, TablePath}, +}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, TopicMetadata}; +use tracing::error; + +use crate::{ + FlussSinkConfig, + schema::{FlussTableLayout, RowContext}, +}; + +pub struct TableWriteResult { + pub insertion_errors: u64, + pub messages_processed: u64, +} + +pub struct FlussWriter { + connection: Option, + config: FlussSinkConfig, +} + +impl Display for FlussWriter { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!(formatter, "FlussWriter") + } +} + +impl fmt::Debug for FlussWriter { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("FlussWriter") + .finish_non_exhaustive() + } +} + +impl FlussWriter { + fn get_connection(&self) -> Result<&FlussConnection, Error> { + self.connection + .as_ref() + .ok_or_else(|| Error::InitError("Fluss connection is not initialized".to_string())) + } + + pub async fn connect(&mut self) -> Result<(), Error> { + let config = fluss::config::Config::try_from(&self.config)?; + let connection = FlussConnection::new(config) + .await + .map_err(|error| Error::InitError(format!("Failed to connect to Fluss: {error}")))?; + // creating writer and cache it + connection.get_or_create_writer_client().map_err(|error| { + Error::InvalidConfigValue(format!("Invalid Fluss writer configuration: {error}")) + })?; + self.connection = Some(connection); + Ok(()) + } + + pub fn new(config: FlussSinkConfig) -> Self { + Self { + config, + connection: None, + } + } + async fn create_table_if_not_exists( + &self, + table_path: &TablePath, + table_descriptor: &TableDescriptor, + ) -> Result<(), Error> { + self.get_connection()? + .get_admin() + .map_err(|error| { + Error::CannotStoreData(format!("Failed to get Fluss admin client: {error}")) + })? + .create_table(table_path, table_descriptor, true) + .await + .map_err(|error| { + Error::CannotStoreData(format!( + "Failed to create Fluss table '{table_path}': {error}" + )) + }) + } + + pub async fn ensure_table_exists( + &self, + table_path: &TablePath, + table_layout: &FlussTableLayout, + ) -> Result<(), Error> { + if self.config.auto_create_table { + let table_descriptor = table_layout.build_table_descriptor().map_err(|error| { + Error::SchemaMismatch(format!("Failed to build Fluss table descriptor: {error}")) + })?; + + self.create_table_if_not_exists(table_path, &table_descriptor) + .await?; + } + Ok(()) + } + + pub async fn write_to_table( + &self, + table_path: &TablePath, + messages_metadata: MessagesMetadata, + messages: Vec, + topic_metadata: &TopicMetadata, + table_layout: &FlussTableLayout, + ) -> Result { + let mut result = TableWriteResult { + insertion_errors: 0, + messages_processed: 0, + }; + + let table = self + .get_connection()? + .get_table(table_path) + .await + .map_err(|error| { + Error::CannotStoreData(format!("Failed to get Fluss table '{table_path}': {error}")) + })?; + + let writer = table + .new_append() + .map_err(|error| { + Error::CannotStoreData(format!( + "Failed to create appender for Fluss table '{table_path}': {error}" + )) + })? + .create_writer() + .map_err(|error| { + Error::CannotStoreData(format!( + "Failed to create writer for Fluss table '{table_path}': {error}" + )) + })?; + + let context = RowContext { + topic: &topic_metadata.topic, + stream: &topic_metadata.stream, + partition_id: messages_metadata.partition_id, + }; + for message in messages { + let row = match table_layout.row_from_message(&message, context) { + Ok(row) => row, + Err(e) => { + error!( + "Can not convert iggy message to row, skipping message id: [{}] because of error: [{}]", + message.id, e + ); + result.insertion_errors += 1; + continue; + } + }; + writer.append(&row).map_err(|error| { + Error::CannotStoreData(format!( + "Failed to append message {} to Fluss table '{table_path}': {error}", + message.id + )) + })?; + result.messages_processed += 1; + } + + writer.flush().await.map_err(|error| { + Error::CannotStoreData(format!( + "Failed to flush rows to Fluss table '{table_path}': {error}" + )) + })?; + + Ok(result) + } +} diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index d4d0bdbd53..3f52c2c714 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -53,6 +53,7 @@ ctor = { workspace = true } deltalake = { workspace = true } dtor = { workspace = true } figment = { workspace = true } +fluss-rs = "0.1.0" futures = { workspace = true } harness_derive = { workspace = true } humantime = { workspace = true } diff --git a/core/integration/tests/connectors/fixtures/fluss/cluster.rs b/core/integration/tests/connectors/fixtures/fluss/cluster.rs new file mode 100644 index 0000000000..75ecaca8de --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/cluster.rs @@ -0,0 +1,228 @@ +// 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 std::fmt::{Display, Formatter}; +use std::net::TcpListener; + +use fluss::client::FlussConnection; +use integration::harness::TestBinaryError; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +use crate::connectors::fixtures; + +const FLUSS_IMAGE: &str = "apache/fluss"; +const ZOOKEEPER_IMAGE: &str = "zookeeper"; +const ZOOKEEPER_VERSION: &str = "3.9.2"; +const ZOOKEEPER_PORT: u16 = 2181; +const FLUSS_CLIENT_PORT: u16 = 9123; +const CONNECTION_RETRY: u16 = 3; +const CONNECTION_RETRY_DELAY_S: u64 = 5; + +struct CoordinatorProperties { + zookeeper_address: String, + container_name: String, + advertised_port: u16, +} + +struct TabletServerProperties { + zookeeper_address: String, + container_name: String, + advertised_port: u16, + tablet_server_id: u32, +} + +impl Display for CoordinatorProperties { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!( + formatter, + "zookeeper.address: {}\n\ + bind.listeners: INTERNAL://{}:0, CLIENT://{}:{}\n\ + advertised.listeners: CLIENT://localhost:{}\n\ + internal.listener.name: INTERNAL\n\ + remote.data.dir: /tmp/fluss/remote-data", + self.zookeeper_address, + self.container_name, + self.container_name, + FLUSS_CLIENT_PORT, + self.advertised_port, + ) + } +} + +impl Display for TabletServerProperties { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!( + formatter, + "zookeeper.address: {}\n\ + bind.listeners: INTERNAL://{}:0, CLIENT://{}:{}\n\ + advertised.listeners: CLIENT://localhost:{}\n\ + internal.listener.name: INTERNAL\n\ + tablet-server.id: {}\n\ + kv.snapshot.interval: 0s\n\ + data.dir: /tmp/fluss/data/tablet-server-{}\n\ + remote.data.dir: /tmp/fluss/remote-data", + self.zookeeper_address, + self.container_name, + self.container_name, + FLUSS_CLIENT_PORT, + self.advertised_port, + self.tablet_server_id, + self.tablet_server_id, + ) + } +} + +pub struct FlussCluster { + #[allow(dead_code)] + zookeeper: ContainerAsync, + #[allow(dead_code)] + coordinator_server: ContainerAsync, + #[allow(dead_code)] + tablet_server: ContainerAsync, + pub coordinator_address: String, + #[allow(dead_code)] + pub fluss_version: String, +} + +impl FlussCluster { + pub async fn new(fluss_version: &str) -> Result { + Self::start(fluss_version).await + } + + pub async fn get_connection(&self) -> Result { + let config = fluss::config::Config { + bootstrap_servers: self.coordinator_address.clone(), + ..fluss::config::Config::default() + }; + + FlussConnection::new(config).await.map_err(|error| { + super::fixture_error(format!("Failed to create Fluss connection: {error}")) + }) + } + + async fn wait_for_fluss_to_become_healthy(&self) -> Result<(), TestBinaryError> { + let mut attempts = 0; + loop { + match self.get_connection().await { + Ok(_) => return Ok(()), + Err(error) => { + attempts += 1; + if attempts >= CONNECTION_RETRY { + return Err(super::fixture_error(format!( + "Failed to establish Fluss connection after {} attempts: {}", + CONNECTION_RETRY, error + ))); + } + tokio::time::sleep(std::time::Duration::from_secs(CONNECTION_RETRY_DELAY_S)) + .await; + } + } + } + } + + async fn start(fluss_version: &str) -> Result { + let network = fixtures::unique_container_name("fluss-network"); + let zookeeper_name = fixtures::unique_container_name("fluss-zookeeper"); + let coordinator_name = fixtures::unique_container_name("fluss-coordinator"); + let tablet_name = fixtures::unique_container_name("fluss-tablet-0"); + let coordinator_host_port = available_host_port()?; + let tablet_host_port = available_host_port_except(coordinator_host_port)?; + + let zookeeper = GenericImage::new(ZOOKEEPER_IMAGE, ZOOKEEPER_VERSION) + .with_exposed_port(ZOOKEEPER_PORT.tcp()) + .with_wait_for(WaitFor::message_on_stdout("Started AdminServer")) + .with_network(&network) + .with_container_name(&zookeeper_name) + .start() + .await + .map_err(|error| super::fixture_error(format!("Failed to start ZooKeeper: {error}")))?; + + let zookeeper_address = format!("{zookeeper_name}:{ZOOKEEPER_PORT}"); + let coordinator_properties = CoordinatorProperties { + zookeeper_address: zookeeper_address.clone(), + container_name: coordinator_name.clone(), + advertised_port: coordinator_host_port, + }; + let coordinator_server = GenericImage::new(FLUSS_IMAGE, fluss_version) + .with_exposed_port(FLUSS_CLIENT_PORT.tcp()) + .with_wait_for(WaitFor::Nothing) + .with_network(&network) + .with_container_name(&coordinator_name) + .with_env_var("FLUSS_PROPERTIES", coordinator_properties.to_string()) + .with_cmd(["coordinatorServer"]) + .with_mapped_port(coordinator_host_port, FLUSS_CLIENT_PORT.tcp()) + .start() + .await + .map_err(|error| { + super::fixture_error(format!("Failed to start Fluss coordinator server: {error}")) + })?; + + let tablet_properties = TabletServerProperties { + zookeeper_address: zookeeper_address.clone(), + container_name: tablet_name.clone(), + advertised_port: tablet_host_port, + tablet_server_id: 0, + }; + let tablet_server = GenericImage::new(FLUSS_IMAGE, fluss_version) + .with_exposed_port(FLUSS_CLIENT_PORT.tcp()) + .with_wait_for(WaitFor::Nothing) + .with_network(&network) + .with_container_name(&tablet_name) + .with_env_var("FLUSS_PROPERTIES", tablet_properties.to_string()) + .with_cmd(["tabletServer"]) + .with_mapped_port(tablet_host_port, FLUSS_CLIENT_PORT.tcp()) + .start() + .await + .map_err(|error| { + super::fixture_error(format!("Failed to start Fluss tablet server: {error}")) + })?; + + let result = Self { + fluss_version: fluss_version.to_string(), + zookeeper, + coordinator_server, + tablet_server, + coordinator_address: format!("localhost:{coordinator_host_port}"), + }; + + result.wait_for_fluss_to_become_healthy().await?; + + Ok(result) + } +} + +fn available_host_port() -> Result { + let listener = TcpListener::bind(("127.0.0.1", 0)) + .map_err(|error| super::fixture_error(format!("Failed to reserve a host port: {error}")))?; + listener + .local_addr() + .map(|address| address.port()) + .map_err(|error| { + super::fixture_error(format!("Failed to read the reserved host port: {error}")) + }) +} + +fn available_host_port_except(excluded_port: u16) -> Result { + loop { + let host_port = available_host_port()?; + if host_port != excluded_port { + return Ok(host_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..16643c2303 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/mod.rs @@ -0,0 +1,29 @@ +// 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 cluster; +mod sink; + +use integration::harness::TestBinaryError; +pub use sink::FlussSinkFixture; + +fn fixture_error(message: String) -> TestBinaryError { + TestBinaryError::FixtureSetup { + fixture_type: "FlussCluster".to_string(), + message, + } +} diff --git a/core/integration/tests/connectors/fixtures/fluss/sink.rs b/core/integration/tests/connectors/fixtures/fluss/sink.rs new file mode 100644 index 0000000000..20561455d3 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/sink.rs @@ -0,0 +1,190 @@ +// 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 std::{collections::HashMap, time::Duration}; + +use crate::connectors::fixtures::fluss::fixture_error; + +use super::cluster::FlussCluster; +use async_trait::async_trait; +use fluss::row::ColumnarRow; +use fluss::{ + client::{EARLIEST_OFFSET, FlussConnection}, + metadata::{TableInfo, TablePath}, +}; +use integration::harness::{TestBinaryError, TestFixture, seeds}; +use tokio::time::{Instant, timeout_at}; + +const DEFAULT_FLUSS_VERSION: &str = "0.9.1-incubating"; +const DEFAULT_SINK_DB: &str = "fluss"; +const DEFAULT_SINK_TABLE: &str = "iggy_messages"; + +const ENV_SINK_BOOTSTRAP_SERVERS: &str = + "IGGY_CONNECTORS_SINK_FLUSS_PLUGIN_CONFIG_BOOTSTRAP_SERVERS"; +const ENV_SINK_TARGET_TABLE: &str = "IGGY_CONNECTORS_SINK_FLUSS_PLUGIN_CONFIG_TARGET_TABLE"; +const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_STREAM"; +const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_TOPICS"; +const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_SCHEMA"; +const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_CONSUMER_GROUP"; +const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_FLUSS_PATH"; + +fn create_test_table_path() -> TablePath { + TablePath::new(DEFAULT_SINK_DB, DEFAULT_SINK_TABLE) +} + +pub struct FlussSinkFixture { + cluster: FlussCluster, +} + +impl FlussSinkFixture { + pub async fn get_fluss_connection(&self) -> Result { + self.cluster.get_connection().await + } + + pub async fn get_test_table(&self) -> Result { + let connection = self.get_fluss_connection().await?; + let admin = connection + .get_admin() + .map_err(|error| fixture_error(format!("Failed to get Fluss admin: {error}")))?; + + admin + .get_table_info(&create_test_table_path()) + .await + .map_err(|error| fixture_error(format!("Failed to get Fluss test table: {error}"))) + } + + pub async fn read_from_test_table( + &self, + timeout: u64, + ) -> Result, TestBinaryError> { + let connection = self.get_fluss_connection().await?; + let table_path = create_test_table_path(); + let table = connection + .get_table(&table_path) + .await + .map_err(|e| fixture_error(format!("Failed to get table: {}", e)))?; + + let log_scanner = table + .new_scan() + .create_log_scanner() + .map_err(|e| fixture_error(format!("Failed to create log scanner: {}", e)))?; + + log_scanner + .subscribe(0, EARLIEST_OFFSET) + .await + .map_err(|e| fixture_error(format!("Failed to subscribe to log scanner: {}", e)))?; + + let deadline = Instant::now() + Duration::from_secs(timeout); + + let mut rows: Vec = Vec::new(); + + loop { + let records = match timeout_at(deadline, log_scanner.poll(Duration::from_secs(5))).await + { + Ok(Err(e)) => { + return Err(fixture_error(format!("Failed to poll log scanner: {}", e))); + } + Ok(Ok(records)) => records, + Err(_) => break, + }; + + for record in records { + rows.push(record.row); + } + } + + Ok(rows) + } + + pub async fn check_if_test_table_exists(&self) -> Result { + let connection = self.get_fluss_connection().await?; + let admin = connection + .get_admin() + .map_err(|e| fixture_error(format!("Error getting Fluss admin instance {}", e)))?; + let exists = admin + .table_exists(&create_test_table_path()) + .await + .map_err(|e| fixture_error(format!("Error checking if table exists {}", e)))?; + Ok(exists) + } + + pub async fn wait_for_test_table(&self, timeout: u64) -> Result<(), TestBinaryError> { + let deadline = Instant::now() + Duration::from_secs(timeout); + loop { + let timeout_at = timeout_at(deadline, self.check_if_test_table_exists()).await; + match timeout_at { + Ok(Ok(false)) => {} + Ok(Ok(true)) => return Ok(()), + Ok(Err(e)) => { + return Err(fixture_error( + format!("Checking the table has failed with: {}", e).to_string(), + )); + } + Err(_) => { + return Err(fixture_error( + format!( + "Test table was not created within the timeout of {}s", + timeout + ) + .to_string(), + )); + } + }; + + tokio::time::sleep(Duration::from_millis(50)).await; + } + } +} + +#[async_trait] +impl TestFixture for FlussSinkFixture { + async fn setup() -> Result { + let cluster = FlussCluster::new(DEFAULT_FLUSS_VERSION).await?; + Ok(Self { cluster }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([ + ( + ENV_SINK_BOOTSTRAP_SERVERS.to_string(), + self.cluster.coordinator_address.clone(), + ), + ( + ENV_SINK_TARGET_TABLE.to_string(), + DEFAULT_SINK_TABLE.to_string(), + ), + ( + ENV_SINK_STREAMS_0_STREAM.to_string(), + seeds::names::STREAM.to_string(), + ), + ( + ENV_SINK_STREAMS_0_TOPICS.to_string(), + format!("[{}]", seeds::names::TOPIC), + ), + (ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string()), + ( + ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(), + seeds::names::CONSUMER_GROUP.to_string(), + ), + ( + ENV_SINK_PATH.to_string(), + "../../target/debug/libiggy_connector_fluss_sink".to_string(), + ), + ]) + } +} diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index 6d75fbf977..70fbf1f5c7 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; @@ -56,6 +57,7 @@ pub use doris::{ DorisSinkPreCreatedFixture, }; pub use elasticsearch::{ElasticsearchSinkFixture, ElasticsearchSourcePreCreatedFixture}; +pub use fluss::FlussSinkFixture; pub use http::{ HttpSinkIndividualFixture, HttpSinkJsonArrayFixture, HttpSinkMultiTopicFixture, HttpSinkNdjsonFixture, HttpSinkNoMetadataFixture, HttpSinkRawFixture, diff --git a/core/integration/tests/connectors/fluss/fluss_sink.rs b/core/integration/tests/connectors/fluss/fluss_sink.rs new file mode 100644 index 0000000000..d9ef7db9fb --- /dev/null +++ b/core/integration/tests/connectors/fluss/fluss_sink.rs @@ -0,0 +1,239 @@ +// 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::FlussSinkFixture; +use crate::connectors::{TestMessage, create_test_messages}; +use bytes::Bytes; +use fluss::metadata::{Column, DataTypes, Schema}; +use fluss::row::InternalRow; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::Identifier; +use iggy_common::MessageClient; +use integration::harness::seeds; +use integration::iggy_harness; + +const TEST_MESSAGE_COUNT: usize = 10; +const ROW_COMPARISON_MESSAGE_COUNT: usize = 3; +const IGGY_STREAM_COLUMN_INDEX: usize = 4; +const IGGY_TOPIC_COLUMN_INDEX: usize = 5; +const PAYLOAD_COLUMN_INDEX: usize = 8; +const WAIT_TIMEOUT_S: u64 = 10; + +fn expected_sink_schema() -> Schema { + Schema::builder() + .with_columns(vec![ + Column::new("id", DataTypes::string()).with_comment("Apache Iggy message ID"), + Column::new("checksum", DataTypes::decimal(20, 0)) + .with_comment("Apache Iggy message checksum"), + Column::new("iggy_offset", DataTypes::decimal(20, 0)) + .with_comment("Apache Iggy message offset"), + Column::new("iggy_timestamp", DataTypes::timestamp_ltz_with_precision(6)) + .with_comment("Apache Iggy message timestamp"), + Column::new("iggy_stream", DataTypes::string()).with_comment("Apache Iggy stream name"), + Column::new("iggy_topic", DataTypes::string()).with_comment("Apache Iggy topic name"), + Column::new("iggy_partition_id", DataTypes::bigint()) + .with_comment("Apache Iggy partition ID"), + Column::new( + "iggy_origin_timestamp", + DataTypes::timestamp_ltz_with_precision(6), + ) + .with_comment("Apache Iggy message origin timestamp"), + Column::new("payload", DataTypes::string()).with_comment("Apache Iggy message payload"), + ]) + .build() + .expect("Expected Fluss sink schema should build") +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_create_test_table_with_expected_schema( + harness: &TestHarness, + fixture: FlussSinkFixture, +) { + let client = harness + .root_client() + .await + .expect("Root client should be available"); + + let stream_id: Identifier = seeds::names::STREAM + .try_into() + .expect("Stream identifier should be valid"); + let topic_id: Identifier = seeds::names::TOPIC + .try_into() + .expect("Topic identifier should be valid"); + + let messages_data = create_test_messages(1); + let mut messages: Vec = messages_data + .iter() + .enumerate() + .map(|(i, msg)| { + let payload = serde_json::to_vec(msg).expect("Failed to serialize message"); + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table should be created"); + let table = fixture + .get_test_table() + .await + .expect("Fluss test table should be available"); + + assert_eq!(table.schema, expected_sink_schema()); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_write_message_to_test_table(harness: &TestHarness, fixture: FlussSinkFixture) { + let client = harness.root_client().await.unwrap(); + + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let messages_data = create_test_messages(TEST_MESSAGE_COUNT); + let mut messages: Vec = messages_data + .iter() + .enumerate() + .map(|(i, msg)| { + let payload = serde_json::to_vec(msg).expect("Failed to serialize message"); + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Table has not found in time"); + + let messages = fixture + .read_from_test_table(WAIT_TIMEOUT_S) + .await + .expect("read messages"); + + assert_eq!(messages.len(), TEST_MESSAGE_COUNT); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_preserve_stream_topic_and_payload_in_rows( + harness: &TestHarness, + fixture: FlussSinkFixture, +) { + let client = harness + .root_client() + .await + .expect("Root client should be available"); + + let stream_id: Identifier = seeds::names::STREAM + .try_into() + .expect("Stream identifier should be valid"); + let topic_id: Identifier = seeds::names::TOPIC + .try_into() + .expect("Topic identifier should be valid"); + + let expected_messages = create_test_messages(ROW_COMPARISON_MESSAGE_COUNT); + let mut messages: Vec = expected_messages + .iter() + .enumerate() + .map(|(index, message)| { + let payload = + serde_json::to_vec(message).expect("Test message payload should serialize"); + IggyMessage::builder() + .id((index + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Iggy message should build") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Messages should be sent"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table should be created"); + let rows = fixture + .read_from_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table rows should be readable"); + + assert_eq!(rows.len(), expected_messages.len()); + + for (row, expected_message) in rows.iter().zip(&expected_messages) { + assert_eq!( + row.get_string(IGGY_STREAM_COLUMN_INDEX) + .expect("Iggy stream column should contain a string"), + seeds::names::STREAM + ); + assert_eq!( + row.get_string(IGGY_TOPIC_COLUMN_INDEX) + .expect("Iggy topic column should contain a string"), + seeds::names::TOPIC + ); + + let payload = row + .get_string(PAYLOAD_COLUMN_INDEX) + .expect("Payload column should contain a string"); + let actual_message: TestMessage = + serde_json::from_str(payload).expect("Payload should contain a test message"); + assert_eq!(&actual_message, expected_message); + } +} diff --git a/core/integration/tests/connectors/fluss/mod.rs b/core/integration/tests/connectors/fluss/mod.rs new file mode 100644 index 0000000000..842231fee8 --- /dev/null +++ b/core/integration/tests/connectors/fluss/mod.rs @@ -0,0 +1,18 @@ +// 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_sink; diff --git a/core/integration/tests/connectors/fluss/sink.toml b/core/integration/tests/connectors/fluss/sink.toml new file mode 100644 index 0000000000..53914b6ab7 --- /dev/null +++ b/core/integration/tests/connectors/fluss/sink.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/sinks/fluss_sink" diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index a1433160b9..73bb6f2ec3 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; @@ -34,7 +35,6 @@ mod runtime; mod s3; mod stdout; mod surrealdb; - use iggy_common::IggyTimestamp; use serde::{Deserialize, Serialize};