This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1f18a8f08 feat(connectors): add S3 sink connector (#3103)
1f18a8f08 is described below
commit 1f18a8f08ebfd5baad1592b294f7fa1b027cd040
Author: Atharva Lade <[email protected]>
AuthorDate: Mon Jun 29 09:28:56 2026 -0500
feat(connectors): add S3 sink connector (#3103)
---
.github/workflows/_build_rust_artifacts.yml | 2 +-
.github/workflows/edge-release.yml | 1 +
Cargo.lock | 349 +++++++++--
Cargo.toml | 1 +
core/connectors/README.md | 1 +
core/connectors/sinks/README.md | 1 +
core/connectors/sinks/s3_sink/Cargo.toml | 52 ++
core/connectors/sinks/s3_sink/README.md | 164 +++++
core/connectors/sinks/s3_sink/config.toml | 50 ++
core/connectors/sinks/s3_sink/src/buffer.rs | 182 ++++++
core/connectors/sinks/s3_sink/src/client.rs | 169 ++++++
core/connectors/sinks/s3_sink/src/formatter.rs | 393 ++++++++++++
core/connectors/sinks/s3_sink/src/lib.rs | 498 ++++++++++++++++
core/connectors/sinks/s3_sink/src/path.rs | 245 ++++++++
core/connectors/sinks/s3_sink/src/sink.rs | 662 +++++++++++++++++++++
core/integration/Cargo.toml | 1 +
core/integration/tests/cli/common/keyring.rs | 1 +
core/integration/tests/cli/common/mod.rs | 1 +
core/integration/tests/connectors/fixtures/mod.rs | 2 +
.../tests/connectors/fixtures/s3/fixture.rs | 308 ++++++++++
.../tests/connectors/fixtures/s3/mod.rs | 20 +
core/integration/tests/connectors/mod.rs | 1 +
core/integration/tests/connectors/s3/mod.rs | 18 +
core/integration/tests/connectors/s3/s3_sink.rs | 196 ++++++
core/integration/tests/connectors/s3/sink.toml | 20 +
.../tests/connectors/s3/sink_rotation.toml | 20 +
26 files changed, 3317 insertions(+), 41 deletions(-)
diff --git a/.github/workflows/_build_rust_artifacts.yml
b/.github/workflows/_build_rust_artifacts.yml
index db6910e26..97e628376 100644
--- a/.github/workflows/_build_rust_artifacts.yml
+++ b/.github/workflows/_build_rust_artifacts.yml
@@ -46,7 +46,7 @@ on:
connector_plugins:
type: string
required: false
- default:
"iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_stdout_sink"
+ default:
"iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_s3_sink,iggy_connector_stdout_sink"
description: "Comma-separated list of connector plugin crates to build
as shared libraries"
outputs:
artifact_name:
diff --git a/.github/workflows/edge-release.yml
b/.github/workflows/edge-release.yml
index 7ca84fc5d..4aaf4b234 100644
--- a/.github/workflows/edge-release.yml
+++ b/.github/workflows/edge-release.yml
@@ -108,6 +108,7 @@ jobs:
- `iggy_connector_postgres_source`
- `iggy_connector_quickwit_sink`
- `iggy_connector_random_source`
+ - `iggy_connector_s3_sink`
- `iggy_connector_stdout_sink`
## Downloads
diff --git a/Cargo.lock b/Cargo.lock
index be96205a9..d1e1c73b1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1267,6 +1267,22 @@ version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
+[[package]]
+name = "attohttpc"
+version = "0.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9"
+dependencies = [
+ "base64",
+ "http 1.4.2",
+ "log",
+ "rustls",
+ "serde",
+ "serde_json",
+ "url",
+ "webpki-roots 1.0.8",
+]
+
[[package]]
name = "autocfg"
version = "1.5.1"
@@ -1368,6 +1384,23 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "aws-creds"
+version = "0.39.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "ca3b85155d265df828f84e53886ed9e427aed979dd8a39f5b8b2162c77e142d7"
+dependencies = [
+ "attohttpc",
+ "home",
+ "log",
+ "quick-xml 0.38.4",
+ "rust-ini",
+ "serde",
+ "thiserror 2.0.18",
+ "time",
+ "url",
+]
+
[[package]]
name = "aws-lc-rs"
version = "1.17.0"
@@ -1390,6 +1423,15 @@ dependencies = [
"fs_extra",
]
+[[package]]
+name = "aws-region"
+version = "0.28.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "838b36c8dc927b6db1b6c6b8f5d05865f2213550b9e83bf92fa99ed6525472c0"
+dependencies = [
+ "thiserror 2.0.18",
+]
+
[[package]]
name = "aws-runtime"
version = "1.7.5"
@@ -1930,7 +1972,7 @@ dependencies = [
"rand 0.10.1",
"serde",
"serde_json",
- "sysinfo",
+ "sysinfo 0.39.5",
"terminal_size",
"tracing",
"uuid",
@@ -2558,6 +2600,15 @@ dependencies = [
"thiserror 2.0.18",
]
+[[package]]
+name = "castaway"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a"
+dependencies = [
+ "rustversion",
+]
+
[[package]]
name = "cbc"
version = "0.1.2"
@@ -2660,7 +2711,7 @@ dependencies = [
"num-traits",
"serde",
"wasm-bindgen",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -2816,6 +2867,19 @@ dependencies = [
"unicode-width 0.2.2",
]
+[[package]]
+name = "compact_str"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f86b9c4c00838774a6d902ef931eff7470720c51d90c2e32cfe15dc304737b3f"
+dependencies = [
+ "castaway",
+ "cfg-if",
+ "itoa",
+ "ryu",
+ "static_assertions",
+]
+
[[package]]
name = "compio"
version = "0.19.1"
@@ -5182,8 +5246,8 @@ dependencies = [
"libc",
"log",
"rustversion",
- "windows-link",
- "windows-result",
+ "windows-link 0.2.1",
+ "windows-result 0.4.1",
]
[[package]]
@@ -6031,7 +6095,7 @@ checksum =
"617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd"
dependencies = [
"cfg-if",
"libc",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -6292,7 +6356,7 @@ dependencies = [
"js-sys",
"log",
"wasm-bindgen",
- "windows-core",
+ "windows-core 0.62.2",
]
[[package]]
@@ -6567,7 +6631,7 @@ dependencies = [
"rand 0.10.1",
"rayon",
"serde",
- "sysinfo",
+ "sysinfo 0.39.5",
"tokio",
"tracing",
"tracing-appender",
@@ -6677,7 +6741,7 @@ dependencies = [
"serde_with",
"serde_yaml_ng",
"strum 0.28.0",
- "sysinfo",
+ "sysinfo 0.39.5",
"tempfile",
"thiserror 2.0.18",
"tokio",
@@ -7019,6 +7083,27 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "iggy_connector_s3_sink"
+version = "0.4.0"
+dependencies = [
+ "async-trait",
+ "base64",
+ "byte-unit",
+ "chrono",
+ "dashmap",
+ "humantime",
+ "iggy_common",
+ "iggy_connector_sdk",
+ "rust-s3",
+ "secrecy",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "tokio",
+ "tracing",
+]
+
[[package]]
name = "iggy_connector_sdk"
version = "0.3.1-edge.1"
@@ -7285,6 +7370,7 @@ dependencies = [
"reqwest-middleware",
"reqwest-retry",
"rmcp",
+ "rust-s3",
"secrecy",
"serde",
"serde_json",
@@ -7292,7 +7378,7 @@ dependencies = [
"server",
"socket2 0.6.4",
"sqlx",
- "sysinfo",
+ "sysinfo 0.39.5",
"tempfile",
"test-case",
"testcontainers",
@@ -7349,7 +7435,7 @@ dependencies = [
"socket2 0.6.4",
"widestring",
"windows-registry",
- "windows-result",
+ "windows-result 0.4.1",
"windows-sys 0.61.2",
]
@@ -7405,7 +7491,7 @@ dependencies = [
"portable-atomic",
"portable-atomic-util",
"serde_core",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -7464,7 +7550,7 @@ dependencies = [
"simd_cesu8",
"thiserror 2.0.18",
"walkdir",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -7779,7 +7865,7 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55"
dependencies = [
"cfg-if",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -8122,6 +8208,17 @@ version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+[[package]]
+name = "maybe-async"
+version = "0.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "746873a384ad60adc5db74471dfaba74bd278afbdcfd81db93fafcdfc8b5ca0c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
[[package]]
name = "maybe-rayon"
version = "0.1.1"
@@ -8152,6 +8249,12 @@ dependencies = [
"digest 0.11.3",
]
+[[package]]
+name = "md5"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "ae960838283323069879657ca3de837e9f7bbb4c7bf6ea7f1b290d5e9476d2e0"
+
[[package]]
name = "memchr"
version = "2.8.2"
@@ -8277,6 +8380,15 @@ dependencies = [
"unicase",
]
+[[package]]
+name = "minidom"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e394a0e3c7ccc2daea3dffabe82f09857b6b510cb25af87d54bf3e910ac1642d"
+dependencies = [
+ "rxml",
+]
+
[[package]]
name = "minimal-lexical"
version = "0.2.1"
@@ -9170,7 +9282,7 @@ dependencies = [
"libc",
"redox_syscall",
"smallvec",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -9297,7 +9409,7 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "9510c76e37ee20be1edfce1819a3169a7c1c67fab98bd35a33684f8886f5cf4d"
dependencies = [
"libc",
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -10857,6 +10969,41 @@ dependencies = [
"ordered-multimap",
]
+[[package]]
+name = "rust-s3"
+version = "0.37.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "aeedb13abdaa7e48d391de05b0569b37fa0a7a64a668dff6ffb2141ad0c2527e"
+dependencies = [
+ "async-trait",
+ "aws-creds",
+ "aws-region",
+ "base64",
+ "bytes",
+ "cfg-if",
+ "futures-util",
+ "hex",
+ "hmac 0.12.1",
+ "http 1.4.2",
+ "log",
+ "maybe-async",
+ "md5",
+ "minidom",
+ "percent-encoding",
+ "quick-xml 0.38.4",
+ "reqwest 0.12.28",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "sha2 0.10.9",
+ "sysinfo 0.37.2",
+ "thiserror 2.0.18",
+ "time",
+ "tokio",
+ "tokio-stream",
+ "url",
+]
+
[[package]]
name = "rust_decimal"
version = "1.42.1"
@@ -11065,6 +11212,25 @@ dependencies = [
"unicode-script",
]
+[[package]]
+name = "rxml"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "65bc94b580d0f5a6b7a2d604e597513d3c673154b52ddeccd1d5c32360d945ee"
+dependencies = [
+ "bytes",
+ "rxml_validation",
+]
+
+[[package]]
+name = "rxml_validation"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "826e80413b9a35e9d33217b3dcac04cf95f6559d15944b93887a08be5496c4a4"
+dependencies = [
+ "compact_str",
+]
+
[[package]]
name = "ryu"
version = "1.0.23"
@@ -11547,7 +11713,7 @@ dependencies = [
"slab",
"socket2 0.6.4",
"strum 0.28.0",
- "sysinfo",
+ "sysinfo 0.39.5",
"tempfile",
"thiserror 2.0.18",
"toml 1.1.2+spec-1.1.0",
@@ -11628,7 +11794,7 @@ dependencies = [
"slab",
"socket2 0.6.4",
"strum 0.28.0",
- "sysinfo",
+ "sysinfo 0.39.5",
"tempfile",
"thiserror 2.0.18",
"tokio",
@@ -12492,6 +12658,20 @@ dependencies = [
"syn 2.0.118",
]
+[[package]]
+name = "sysinfo"
+version = "0.37.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "16607d5caffd1c07ce073528f9ed972d88db15dd44023fa57142963be3feb11f"
+dependencies = [
+ "libc",
+ "memchr",
+ "ntapi",
+ "objc2-core-foundation",
+ "objc2-io-kit",
+ "windows 0.61.3",
+]
+
[[package]]
name = "sysinfo"
version = "0.39.5"
@@ -12504,7 +12684,7 @@ dependencies = [
"objc2-core-foundation",
"objc2-io-kit",
"objc2-open-directory",
- "windows",
+ "windows 0.62.2",
]
[[package]]
@@ -13790,7 +13970,7 @@ dependencies = [
"bon",
"rustc_version",
"rustversion",
- "sysinfo",
+ "sysinfo 0.39.5",
"time",
"vergen-lib",
]
@@ -14096,16 +14276,38 @@ version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+[[package]]
+name = "windows"
+version = "0.61.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893"
+dependencies = [
+ "windows-collections 0.2.0",
+ "windows-core 0.61.2",
+ "windows-future 0.2.1",
+ "windows-link 0.1.3",
+ "windows-numerics 0.2.0",
+]
+
[[package]]
name = "windows"
version = "0.62.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580"
dependencies = [
- "windows-collections",
- "windows-core",
- "windows-future",
- "windows-numerics",
+ "windows-collections 0.3.2",
+ "windows-core 0.62.2",
+ "windows-future 0.3.2",
+ "windows-numerics 0.3.1",
+]
+
+[[package]]
+name = "windows-collections"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8"
+dependencies = [
+ "windows-core 0.61.2",
]
[[package]]
@@ -14114,7 +14316,20 @@ version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610"
dependencies = [
- "windows-core",
+ "windows-core 0.62.2",
+]
+
+[[package]]
+name = "windows-core"
+version = "0.61.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
+dependencies = [
+ "windows-implement",
+ "windows-interface",
+ "windows-link 0.1.3",
+ "windows-result 0.3.4",
+ "windows-strings 0.4.2",
]
[[package]]
@@ -14125,9 +14340,20 @@ checksum =
"b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
dependencies = [
"windows-implement",
"windows-interface",
- "windows-link",
- "windows-result",
- "windows-strings",
+ "windows-link 0.2.1",
+ "windows-result 0.4.1",
+ "windows-strings 0.5.1",
+]
+
+[[package]]
+name = "windows-future"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+ "windows-threading 0.1.0",
]
[[package]]
@@ -14136,9 +14362,9 @@ version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb"
dependencies = [
- "windows-core",
- "windows-link",
- "windows-threading",
+ "windows-core 0.62.2",
+ "windows-link 0.2.1",
+ "windows-threading 0.2.1",
]
[[package]]
@@ -14163,6 +14389,12 @@ dependencies = [
"syn 2.0.118",
]
+[[package]]
+name = "windows-link"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
+
[[package]]
name = "windows-link"
version = "0.2.1"
@@ -14182,14 +14414,24 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "windows-numerics"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+]
+
[[package]]
name = "windows-numerics"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26"
dependencies = [
- "windows-core",
- "windows-link",
+ "windows-core 0.62.2",
+ "windows-link 0.2.1",
]
[[package]]
@@ -14198,9 +14440,18 @@ version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720"
dependencies = [
- "windows-link",
- "windows-result",
- "windows-strings",
+ "windows-link 0.2.1",
+ "windows-result 0.4.1",
+ "windows-strings 0.5.1",
+]
+
+[[package]]
+name = "windows-result"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
+dependencies = [
+ "windows-link 0.1.3",
]
[[package]]
@@ -14209,7 +14460,16 @@ version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
+dependencies = [
+ "windows-link 0.1.3",
]
[[package]]
@@ -14218,7 +14478,7 @@ version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -14263,7 +14523,7 @@ version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
@@ -14303,7 +14563,7 @@ version = "0.53.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
"windows_aarch64_gnullvm 0.53.1",
"windows_aarch64_msvc 0.53.1",
"windows_i686_gnu 0.53.1",
@@ -14314,13 +14574,22 @@ dependencies = [
"windows_x86_64_msvc 0.53.1",
]
+[[package]]
+name = "windows-threading"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6"
+dependencies = [
+ "windows-link 0.1.3",
+]
+
[[package]]
name = "windows-threading"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37"
dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 4d196d963..027209cea 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -42,6 +42,7 @@ members = [
"core/connectors/sinks/mongodb_sink",
"core/connectors/sinks/postgres_sink",
"core/connectors/sinks/quickwit_sink",
+ "core/connectors/sinks/s3_sink",
"core/connectors/sinks/stdout_sink",
"core/connectors/sources/elasticsearch_source",
"core/connectors/sources/influxdb_source",
diff --git a/core/connectors/README.md b/core/connectors/README.md
index b7d24cfba..2d26715fa 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -85,6 +85,7 @@ Each sink should have its own, custom configuration, which is
passed along with
- **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
- **PostgreSQL Sink** - stores messages in PostgreSQL database tables
- **Quickwit Sink** - indexes messages in Quickwit search engine
+- **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO,
R2, B2, DO Spaces)
- **Stdout Sink** - prints messages to standard output (useful for
debugging/development)
## Source
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 367a22028..497617f22 100644
--- a/core/connectors/sinks/README.md
+++ b/core/connectors/sinks/README.md
@@ -14,6 +14,7 @@ Sink connectors are responsible for writing data from Iggy
streams to external s
| **influxdb_sink** | Writes messages to InfluxDB as line-protocol points;
supports both V2 (org/bucket, Flux) and V3 (db, SQL) |
| **postgres_sink** | Stores messages in PostgreSQL database tables with
configurable schemas |
| **quickwit_sink** | Indexes messages in Quickwit search engine for log
analytics |
+| **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO,
R2, B2, DO Spaces) |
| **stdout_sink** | Prints messages to standard output (useful for debugging
and development) |
The sink is represented by the single `Sink` trait, which defines the basic
interface for all sink connectors. It provides methods for initializing the
sink, writing data to external destination, and closing the sink.
diff --git a/core/connectors/sinks/s3_sink/Cargo.toml
b/core/connectors/sinks/s3_sink/Cargo.toml
new file mode 100644
index 000000000..b70647413
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/Cargo.toml
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_s3_sink"
+version = "0.4.0"
+description = "Iggy S3 sink connector for writing stream messages to Amazon S3
and S3-compatible stores"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "s3", "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 }
+base64 = { workspace = true }
+byte-unit = { workspace = true }
+chrono = { workspace = true }
+dashmap = { workspace = true }
+humantime = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+rust-s3 = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+simd-json = { workspace = true }
diff --git a/core/connectors/sinks/s3_sink/README.md
b/core/connectors/sinks/s3_sink/README.md
new file mode 100644
index 000000000..6ee55368c
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/README.md
@@ -0,0 +1,164 @@
+# Apache Iggy S3 Sink Connector
+
+Writes messages from Iggy streams to Amazon S3 and S3-compatible object stores
(MinIO, Cloudflare R2, DigitalOcean Spaces, Backblaze B2).
+
+## Features
+
+- Buffered uploads with configurable file rotation (by size or message count)
+- Multiple output formats: JSON Lines, JSON Array, Raw
+- Configurable path templates with variables for stream, topic, date, hour,
partition
+- S3 keys include offset ranges for human-readable object naming
+- Optional metadata and header inclusion in output
+- Support for custom endpoints (MinIO, R2) and path-style addressing
+- Retry with exponential backoff and jitter on transient upload failures
+
+## Configuration
+
+### Connector Runtime Config
+
+```toml
+type = "sink"
+key = "s3"
+enabled = true
+version = 0
+name = "S3 sink"
+path = "../../target/release/libiggy_connector_s3_sink"
+verbose = false
+
+[[streams]]
+stream = "application_logs"
+topics = ["api_requests", "errors"]
+schema = "json"
+batch_length = 1000
+poll_interval = "100ms"
+consumer_group = "s3_sink"
+```
+
+### Plugin Configuration
+
+```toml
+[plugin_config]
+bucket = "my-data-lake"
+prefix = "iggy/raw"
+region = "us-east-1"
+# endpoint = "http://localhost:9000"; # for MinIO / S3-compatible stores
+# access_key_id = "AKIA..." # omit to use env vars / instance
profile
+# secret_access_key = "..." # omit to use env vars / instance
profile
+path_template = "{stream}/{topic}/{date}/{hour}"
+file_rotation = "size"
+max_file_size = "8MiB"
+output_format = "json_lines"
+include_metadata = true
+include_headers = true
+max_attempts = 3
+retry_delay = "1s"
+```
+
+### Options Reference
+
+| Option | Type | Default | Description |
+| ------ | ---- | ------- | ----------- |
+| `bucket` | String | **required** | S3 bucket name |
+| `region` | String | **required** | AWS region (e.g. `us-east-1`) |
+| `prefix` | String | `None` | Key prefix prepended to all objects |
+| `endpoint` | String | `None` | Custom S3 endpoint for MinIO, R2, etc. |
+| `access_key_id` | String | `None` | AWS access key; omit for env/instance
profile |
+| `secret_access_key` | String | `None` | AWS secret key; omit for
env/instance profile |
+| `path_template` | String | `{stream}/{topic}/{date}/{hour}` | Template for
S3 key directory structure |
+| `file_rotation` | String | `size` | Rotation strategy: `size` or `messages` |
+| `max_file_size` | String | `8MiB` | Max file size before rotation (when
`file_rotation = "size"`) |
+| `max_messages_per_file` | Integer | `None` | Max messages per file (required
when `file_rotation = "messages"`) |
+| `output_format` | String | `json_lines` | Output format: `json_lines`,
`json_array`, or `raw` |
+| `include_metadata` | Boolean | `true` | Include
stream/topic/partition/offset in output |
+| `include_headers` | Boolean | `false` | Include message headers in output |
+| `max_attempts` | Integer | `3` | Max total upload attempts per file (also
accepts `max_retries` as alias) |
+| `retry_delay` | String | `1s` | Base delay for exponential backoff
(humantime format) |
+| `path_style` | Boolean | auto | Force path-style S3 addressing; auto-enabled
when `endpoint` is set |
+
+### Path Template Variables
+
+| Variable | Description | Example |
+| -------- | ----------- | ------- |
+| `{stream}` | Iggy stream name | `application_logs` |
+| `{topic}` | Iggy topic name | `api_requests` |
+| `{partition}` | Partition ID | `1` |
+| `{date}` | UTC date from first message in buffer | `2026-03-16` |
+| `{hour}` | UTC hour from first message in buffer | `14` |
+| `{timestamp}` | Epoch millis derived from first message timestamp in buffer
(deterministic) | `1710597600000` |
+
+**Note:** `{timestamp}`, `{date}`, and `{hour}` are all derived from the first
message timestamp in each buffer. They are deterministic across retries within
the same process. However, a process restart resets in-memory buffers, so batch
boundaries (and therefore timestamps in the key) may differ after recovery.
+
+### Credentials
+
+Credentials can be provided in three ways (in order of precedence):
+
+1. **Explicit config**: Set both `access_key_id` and `secret_access_key`
+2. **Environment variables**: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`,
`AWS_SESSION_TOKEN`
+3. **Instance profile / IAM role**: Automatic when running on EC2/ECS/EKS
+
+Both `access_key_id` and `secret_access_key` must be provided together or both
omitted.
+
+## Output Example
+
+With `output_format = "json_lines"` and `include_metadata = true`, writing
`api_requests` messages produces:
+
+```text
+s3://my-data-lake/iggy/raw/application_logs/api_requests/2026-03-16/14/000000-000999.jsonl
+```
+
+Each line:
+
+```json
+{"offset":42,"timestamp":"2026-03-16T14:02:31Z","stream":"application_logs","topic":"api_requests","partition_id":1,"payload":{"method":"GET","path":"/api/users","status":200}}
+```
+
+## S3-Compatible Stores
+
+### MinIO
+
+```toml
+[plugin_config]
+bucket = "my-bucket"
+region = "us-east-1"
+endpoint = "http://localhost:9000";
+access_key_id = "minioadmin"
+secret_access_key = "minioadmin"
+```
+
+### Cloudflare R2
+
+```toml
+[plugin_config]
+bucket = "my-bucket"
+region = "auto"
+endpoint = "https://<account-id>.r2.cloudflarestorage.com"
+access_key_id = "..."
+secret_access_key = "..."
+```
+
+## Delivery Semantics
+
+All retry logic lives inside `consume()`. The connector runtime invokes
`consume()` via an FFI callback that returns an `i32` status code. The runtime
does not inspect this return value (see `process_messages()` in
`runtime/src/sink.rs`), so errors logged by the sink are not propagated to the
runtime's retry or alerting mechanisms. Additionally, consumer group offsets
are committed before processing ([runtime issue #1](#known-limitations)). This
means:
+
+- Failed messages are **not retried by the runtime** — only by the sink's
internal retry loop
+- Messages are committed **before delivery** — a crash after commit but before
delivery loses messages
+
+The effective delivery guarantee is **at-most-once** at the runtime level. The
sink's internal retries provide best-effort delivery within each `consume()`
call.
+
+## Known Limitations
+
+1. **Runtime ignores `consume()` status**: The connector runtime invokes
`consume()` via an FFI callback returning `i32`. The `process_messages()`
function in `runtime/src/sink.rs` does not inspect the return value. Errors are
logged internally by the sink but do not trigger runtime-level retry or
alerting. ([#2927](https://github.com/apache/iggy/issues/2927))
+
+2. **Offsets committed before processing**: The `PollingMessages` auto-commit
strategy commits consumer group offsets before `consume()` is called. Combined
with limitation 1, at-least-once delivery is not achievable.
([#2928](https://github.com/apache/iggy/issues/2928))
+
+3. **In-memory buffering only**: There is no write-ahead log. A process crash
loses all in-memory buffered messages that have not yet been flushed to S3.
+
+4. **No dead letter queue**: Failed messages are logged at `error!` level but
not persisted to a DLQ. DLQ support would be a runtime-level feature.
+
+## Building
+
+```bash
+cargo build --release -p iggy_connector_s3_sink
+```
+
+The compiled plugin will be at
`target/release/libiggy_connector_s3_sink.{so,dylib,dll}`.
diff --git a/core/connectors/sinks/s3_sink/config.toml
b/core/connectors/sinks/s3_sink/config.toml
new file mode 100644
index 000000000..ed71fc900
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/config.toml
@@ -0,0 +1,50 @@
+# 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 = "s3"
+enabled = true
+version = 0
+name = "S3 sink"
+path = "../../target/release/libiggy_connector_s3_sink"
+verbose = false
+
+[[streams]]
+stream = "application_logs"
+topics = ["api_requests", "errors"]
+schema = "json"
+batch_length = 1000
+poll_interval = "100ms"
+consumer_group = "s3_sink"
+
+[plugin_config]
+bucket = "my-data-lake"
+prefix = "iggy/raw"
+region = "us-east-1"
+# endpoint = "http://localhost:9000"; # uncomment for MinIO /
S3-compatible stores
+# access_key_id = "minioadmin" # omit to use env vars / instance
profile
+# secret_access_key = "minioadmin" # omit to use env vars / instance
profile
+path_template = "{stream}/{topic}/{date}/{hour}"
+file_rotation = "size"
+max_file_size = "8MiB"
+# max_messages_per_file = 10000 # used when file_rotation =
"messages"
+output_format = "json_lines"
+include_metadata = true
+include_headers = true
+max_attempts = 3
+retry_delay = "1s"
+# path_style = true # auto-enabled when endpoint is set
(for MinIO)
diff --git a/core/connectors/sinks/s3_sink/src/buffer.rs
b/core/connectors/sinks/s3_sink/src/buffer.rs
new file mode 100644
index 000000000..ad49e2651
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/buffer.rs
@@ -0,0 +1,182 @@
+// 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::FileRotation;
+
+#[derive(Debug)]
+pub(crate) struct FileBuffer {
+ data: Vec<u8>,
+ boundaries: Vec<usize>,
+ message_count: u64,
+ first_offset: Option<u64>,
+ last_offset: Option<u64>,
+ first_timestamp_micros: u64,
+}
+
+impl FileBuffer {
+ pub fn new() -> Self {
+ FileBuffer {
+ data: Vec::new(),
+ boundaries: Vec::new(),
+ message_count: 0,
+ first_offset: None,
+ last_offset: None,
+ first_timestamp_micros: 0,
+ }
+ }
+
+ pub fn append(&mut self, entry: &[u8], offset: u64, timestamp_micros: u64)
{
+ self.data.extend_from_slice(entry);
+ self.boundaries.push(self.data.len());
+ self.message_count += 1;
+
+ if self.first_offset.is_none() {
+ self.first_offset = Some(offset);
+ self.first_timestamp_micros = timestamp_micros;
+ }
+ self.last_offset = Some(offset);
+ }
+
+ pub fn should_rotate(&self, rotation: FileRotation, max_size: u64,
max_messages: u64) -> bool {
+ match rotation {
+ FileRotation::Size => self.data.len() as u64 >= max_size,
+ FileRotation::Messages => self.message_count >= max_messages,
+ }
+ }
+
+ pub fn is_empty(&self) -> bool {
+ self.boundaries.is_empty()
+ }
+
+ /// Returns an iterator over individual entry slices without copying.
+ pub fn entries(&self) -> impl Iterator<Item = &[u8]> {
+ let mut start = 0;
+ self.boundaries.iter().map(move |&end| {
+ let slice = &self.data[start..end];
+ start = end;
+ slice
+ })
+ }
+
+ pub fn first_offset(&self) -> u64 {
+ self.first_offset.unwrap_or(0)
+ }
+
+ pub fn last_offset(&self) -> u64 {
+ self.last_offset.unwrap_or(0)
+ }
+
+ pub fn first_timestamp_micros(&self) -> u64 {
+ self.first_timestamp_micros
+ }
+
+ pub fn message_count(&self) -> u64 {
+ self.message_count
+ }
+
+ pub fn reset(&mut self) {
+ self.data.clear();
+ self.boundaries.clear();
+ self.message_count = 0;
+ self.first_offset = None;
+ self.last_offset = None;
+ self.first_timestamp_micros = 0;
+ }
+}
+
+impl Default for FileBuffer {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn new_buffer_is_empty() {
+ let buf = FileBuffer::new();
+ assert!(buf.is_empty());
+ assert_eq!(buf.message_count(), 0);
+ assert_eq!(buf.first_offset(), 0);
+ assert_eq!(buf.last_offset(), 0);
+ }
+
+ #[test]
+ fn append_tracks_offsets() {
+ let mut buf = FileBuffer::new();
+ buf.append(&[1, 2, 3], 10, 1000);
+ buf.append(&[4, 5], 11, 1001);
+ buf.append(&[6], 12, 1002);
+
+ assert!(!buf.is_empty());
+ assert_eq!(buf.message_count(), 3);
+ assert_eq!(buf.first_offset(), 10);
+ assert_eq!(buf.last_offset(), 12);
+ assert_eq!(buf.first_timestamp_micros(), 1000);
+ assert_eq!(buf.entries().count(), 3);
+ }
+
+ #[test]
+ fn rotation_by_size() {
+ let mut buf = FileBuffer::new();
+ buf.append(&[0; 500], 0, 100);
+ assert!(!buf.should_rotate(FileRotation::Size, 1000, 0));
+
+ buf.append(&[0; 500], 1, 200);
+ assert!(buf.should_rotate(FileRotation::Size, 1000, 0));
+
+ buf.append(&[0; 100], 2, 300);
+ assert!(buf.should_rotate(FileRotation::Size, 1000, 0));
+ }
+
+ #[test]
+ fn rotation_by_messages() {
+ let mut buf = FileBuffer::new();
+ buf.append(&[1], 0, 100);
+ buf.append(&[2], 1, 200);
+ assert!(!buf.should_rotate(FileRotation::Messages, 0, 3));
+
+ buf.append(&[3], 2, 300);
+ assert!(buf.should_rotate(FileRotation::Messages, 0, 3));
+ }
+
+ #[test]
+ fn reset_clears_state() {
+ let mut buf = FileBuffer::new();
+ buf.append(&[1, 2, 3], 5, 1000);
+ buf.append(&[4, 5, 6], 6, 2000);
+
+ buf.reset();
+
+ assert!(buf.is_empty());
+ assert_eq!(buf.message_count(), 0);
+ assert_eq!(buf.first_offset(), 0);
+ assert_eq!(buf.last_offset(), 0);
+ }
+
+ #[test]
+ fn contiguous_entries() {
+ let mut buf = FileBuffer::new();
+ buf.append(b"hello", 0, 100);
+ buf.append(b"world", 1, 200);
+
+ let entries: Vec<&[u8]> = buf.entries().collect();
+ assert_eq!(entries, vec![b"hello".as_slice(), b"world".as_slice()]);
+ }
+}
diff --git a/core/connectors/sinks/s3_sink/src/client.rs
b/core/connectors/sinks/s3_sink/src/client.rs
new file mode 100644
index 000000000..3ec489f1b
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/client.rs
@@ -0,0 +1,169 @@
+// 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::S3SinkConfig;
+use iggy_connector_sdk::Error;
+use s3::creds::Credentials;
+use s3::{Bucket, Region};
+use secrecy::ExposeSecret;
+use tracing::info;
+
+pub(crate) async fn create_bucket(config: &S3SinkConfig) ->
Result<Box<Bucket>, Error> {
+ if config.access_key_id.is_some() != config.secret_access_key.is_some() {
+ return Err(Error::InvalidConfigValue(
+ "Partially configured credentials. You must provide both
access_key_id \
+ and secret_access_key, or omit both."
+ .to_owned(),
+ ));
+ }
+
+ let credentials = match (&config.access_key_id, &config.secret_access_key)
{
+ (Some(key), Some(secret)) => {
+ let key_len = key.expose_secret().len();
+ info!("Using explicit S3 credentials (key length: {key_len}
chars)");
+ Credentials::new(
+ Some(key.expose_secret()),
+ Some(secret.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| Error::InitError(format!("Failed to create S3
credentials: {e}")))?
+ }
+ _ => {
+ info!(
+ "No explicit credentials provided, using default credential
chain (env vars / instance profile)"
+ );
+ Credentials::default().map_err(|e| {
+ Error::InitError(format!("Failed to load default S3
credentials: {e}"))
+ })?
+ }
+ };
+
+ let region = match &config.endpoint {
+ Some(endpoint) => {
+ info!("Using custom S3 endpoint: {endpoint}");
+ Region::Custom {
+ region: config.region.clone(),
+ endpoint: endpoint.clone(),
+ }
+ }
+ None => config.region.parse::<Region>().map_err(|e| {
+ Error::InvalidConfigValue(format!("Invalid S3 region '{}': {e}",
config.region))
+ })?,
+ };
+
+ let mut bucket = Bucket::new(&config.bucket, region, credentials)
+ .map_err(|e| Error::InitError(format!("Failed to create S3 bucket
handle: {e}")))?;
+
+ let use_path_style =
config.path_style.unwrap_or(config.endpoint.is_some());
+ if use_path_style {
+ bucket.set_path_style();
+ }
+
+ Ok(bucket)
+}
+
+/// Verify bucket connectivity with a zero-byte probe write. Only requires
+/// `s3:PutObject` permission, unlike `list_page` which needs `s3:ListBucket`
+/// and fails for write-only IAM policies.
+pub(crate) async fn verify_bucket(bucket: &Bucket) -> Result<(), Error> {
+ const PROBE_KEY: &str = ".iggy-sink-probe";
+ bucket.put_object(PROBE_KEY, &[]).await.map_err(|e| {
+ Error::InitError(format!(
+ "S3 bucket '{}' connectivity check failed: {e}",
+ bucket.name
+ ))
+ })?;
+ let _ = bucket.delete_object(PROBE_KEY).await;
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::{
+ FileRotation, default_max_file_size, default_output_format,
default_path_template,
+ };
+
+ fn base_config() -> S3SinkConfig {
+ S3SinkConfig {
+ bucket: "test".to_string(),
+ region: "us-east-1".to_string(),
+ prefix: None,
+ endpoint: None,
+ access_key_id: None,
+ secret_access_key: None,
+ path_template: default_path_template(),
+ file_rotation: FileRotation::Size,
+ max_file_size: default_max_file_size(),
+ max_messages_per_file: None,
+ output_format: default_output_format(),
+ include_metadata: true,
+ include_headers: false,
+ max_attempts: None,
+ retry_delay: None,
+ path_style: None,
+ }
+ }
+
+ #[test]
+ fn validate_both_credentials_present() {
+ let config = S3SinkConfig {
+ access_key_id: Some("AKIAIOSFODNN7EXAMPLE".into()),
+ secret_access_key:
Some("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".into()),
+ ..base_config()
+ };
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ // create_bucket will succeed on credential validation but may fail on
+ // actual region parsing in test env -- we just check it gets past
validation
+ let result = rt.block_on(create_bucket(&config));
+ assert!(result.is_ok() || !format!("{:?}", result).contains("Partially
configured"));
+ }
+
+ #[test]
+ fn validate_no_credentials() {
+ let config = base_config();
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ let result = rt.block_on(create_bucket(&config));
+ assert!(result.is_ok() || !format!("{:?}", result).contains("Partially
configured"));
+ }
+
+ #[test]
+ fn validate_partial_access_key_only() {
+ let config = S3SinkConfig {
+ access_key_id: Some("AKIAIOSFODNN7EXAMPLE".into()),
+ secret_access_key: None,
+ ..base_config()
+ };
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ let result = rt.block_on(create_bucket(&config));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn validate_partial_secret_key_only() {
+ let config = S3SinkConfig {
+ access_key_id: None,
+ secret_access_key: Some("secret".into()),
+ ..base_config()
+ };
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ let result = rt.block_on(create_bucket(&config));
+ assert!(result.is_err());
+ }
+}
diff --git a/core/connectors/sinks/s3_sink/src/formatter.rs
b/core/connectors/sinks/s3_sink/src/formatter.rs
new file mode 100644
index 000000000..bec194c90
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/formatter.rs
@@ -0,0 +1,393 @@
+// 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::OutputFormat;
+use chrono::{DateTime, Utc};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Payload, TopicMetadata,
owned_value_to_serde_json,
+};
+use serde::Serialize;
+use serde_json::Value;
+
+#[derive(Serialize)]
+struct JsonMessage<'a> {
+ #[serde(skip_serializing_if = "Option::is_none")]
+ offset: Option<u64>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ timestamp: Option<&'a str>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ stream: Option<&'a str>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ topic: Option<&'a str>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ partition_id: Option<u32>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ headers: Option<Value>,
+ payload: Value,
+}
+
+pub(crate) fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Result<Vec<u8>, Error> {
+ match format {
+ OutputFormat::JsonLines | OutputFormat::JsonArray =>
format_json_message(
+ message,
+ topic_metadata,
+ messages_metadata,
+ include_metadata,
+ include_headers,
+ ),
+ OutputFormat::Raw => format_raw_message(message),
+ }
+}
+
+fn format_json_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+) -> Result<Vec<u8>, Error> {
+ let ts_str = if include_metadata {
+ Some(timestamp_to_rfc3339(message.timestamp))
+ } else {
+ None
+ };
+ let msg = JsonMessage {
+ offset: if include_metadata {
+ Some(message.offset)
+ } else {
+ None
+ },
+ timestamp: ts_str.as_deref(),
+ stream: if include_metadata {
+ Some(&topic_metadata.stream)
+ } else {
+ None
+ },
+ topic: if include_metadata {
+ Some(&topic_metadata.topic)
+ } else {
+ None
+ },
+ partition_id: if include_metadata {
+ Some(messages_metadata.partition_id)
+ } else {
+ None
+ },
+ headers: if include_headers {
+ message.headers.as_ref().map(serialize_headers)
+ } else {
+ None
+ },
+ payload: payload_to_json_value(&message.payload),
+ };
+
+ serde_json::to_vec(&msg).map_err(|e| {
+ Error::CannotStoreData(format!(
+ "Failed to serialize message at offset {}: {e}",
+ message.offset
+ ))
+ })
+}
+
+fn format_raw_message(message: &ConsumedMessage) -> Result<Vec<u8>, Error> {
+ message.payload.try_to_bytes().map_err(|e| {
+ Error::CannotStoreData(format!(
+ "Failed to extract raw bytes at offset {}: {e}",
+ message.offset
+ ))
+ })
+}
+
+fn serialize_headers(
+ headers: &std::collections::BTreeMap<iggy_common::HeaderKey,
iggy_common::HeaderValue>,
+) -> Value {
+ use iggy_common::HeaderKind;
+ use serde_json::Map;
+
+ let mut obj = Map::new();
+ for (key, value) in headers {
+ let key_str = key.as_str().unwrap_or("").to_string();
+ let json_value = match value.kind() {
+ HeaderKind::String => {
+
Value::String(String::from_utf8_lossy(&value.value()).into_owned())
+ }
+ HeaderKind::Raw => Value::String(base64_encode(&value.value())),
+ HeaderKind::Bool => {
+ let b = !value.value().is_empty() && value.value()[0] != 0;
+ Value::Bool(b)
+ }
+ HeaderKind::Int8 | HeaderKind::Int16 | HeaderKind::Int32 |
HeaderKind::Int64 => {
+ let s = value.to_string_value();
+ s.parse::<i64>()
+ .map(|n| Value::Number(n.into()))
+ .unwrap_or(Value::String(s))
+ }
+ HeaderKind::Uint8 | HeaderKind::Uint16 | HeaderKind::Uint32 |
HeaderKind::Uint64 => {
+ let s = value.to_string_value();
+ s.parse::<u64>()
+ .map(|n| Value::Number(n.into()))
+ .unwrap_or(Value::String(s))
+ }
+ HeaderKind::Float32 | HeaderKind::Float64 => {
+ let s = value.to_string_value();
+ s.parse::<f64>()
+ .ok()
+ .and_then(serde_json::Number::from_f64)
+ .map(Value::Number)
+ .unwrap_or(Value::String(s))
+ }
+ _ => Value::String(value.to_string_value()),
+ };
+ obj.insert(key_str, json_value);
+ }
+ Value::Object(obj)
+}
+
+fn payload_to_json_value(payload: &Payload) -> Value {
+ match payload {
+ Payload::Json(value) => owned_value_to_serde_json(value),
+ Payload::Text(text) => Value::String(text.clone()),
+ Payload::Raw(bytes) => match serde_json::from_slice(bytes) {
+ Ok(v) => v,
+ Err(_) => Value::String(base64_encode(bytes)),
+ },
+ Payload::Proto(text) => Value::String(text.clone()),
+ Payload::FlatBuffer(bytes) => Value::String(base64_encode(bytes)),
+ Payload::Avro(bytes) => Value::String(base64_encode(bytes)),
+ }
+}
+
+fn base64_encode(bytes: &[u8]) -> String {
+ use base64::Engine;
+ base64::engine::general_purpose::STANDARD.encode(bytes)
+}
+
+fn timestamp_to_rfc3339(micros: u64) -> String {
+ let secs = (micros / 1_000_000) as i64;
+ let nanos = ((micros % 1_000_000) * 1_000) as u32;
+ DateTime::<Utc>::from_timestamp(secs, nanos)
+ .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
+ .unwrap_or_else(|| "1970-01-01T00:00:00Z".to_string())
+}
+
+/// Finalize buffer entries into the output byte format.
+pub(crate) fn finalize_buffer<'a>(
+ entries: impl Iterator<Item = &'a [u8]>,
+ format: OutputFormat,
+) -> Vec<u8> {
+ match format {
+ OutputFormat::JsonLines => {
+ let mut result = Vec::new();
+ for entry in entries {
+ result.extend_from_slice(entry);
+ result.push(b'\n');
+ }
+ result
+ }
+ OutputFormat::Raw => {
+ let mut result = Vec::new();
+ for entry in entries {
+ result.extend_from_slice(entry);
+ }
+ result
+ }
+ OutputFormat::JsonArray => {
+ let mut result = Vec::new();
+ result.push(b'[');
+ let mut first = true;
+ for entry in entries {
+ if !first {
+ result.push(b',');
+ }
+ result.extend_from_slice(entry);
+ first = false;
+ }
+ result.push(b']');
+ result
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_connector_sdk::Schema;
+ use std::collections::BTreeMap;
+
+ fn make_json_payload(json_str: &str) -> Payload {
+ let mut bytes = json_str.as_bytes().to_vec();
+ let value = simd_json::to_owned_value(&mut bytes).unwrap();
+ Payload::Json(value)
+ }
+
+ fn make_message(offset: u64, payload: Payload) -> ConsumedMessage {
+ ConsumedMessage {
+ id: 1,
+ offset,
+ checksum: 12345,
+ timestamp: 1_710_597_751_000_000,
+ origin_timestamp: 1_710_597_751_000_000,
+ headers: None,
+ payload,
+ }
+ }
+
+ fn make_topic_metadata() -> TopicMetadata {
+ TopicMetadata {
+ stream: "app_logs".to_string(),
+ topic: "api_requests".to_string(),
+ }
+ }
+
+ fn make_messages_metadata() -> MessagesMetadata {
+ MessagesMetadata {
+ partition_id: 1,
+ current_offset: 42,
+ schema: Schema::Json,
+ }
+ }
+
+ #[test]
+ fn json_lines_with_metadata() {
+ let payload = make_json_payload(r#"{"method":"GET","status":200}"#);
+ let msg = make_message(42, payload);
+ let topic = make_topic_metadata();
+ let meta = make_messages_metadata();
+
+ let bytes =
+ format_message(&msg, &topic, &meta, true, false,
OutputFormat::JsonLines).unwrap();
+ let value: Value = serde_json::from_slice(&bytes).unwrap();
+
+ assert_eq!(value["offset"], 42);
+ assert_eq!(value["stream"], "app_logs");
+ assert_eq!(value["topic"], "api_requests");
+ assert_eq!(value["partition_id"], 1);
+ assert_eq!(value["payload"]["method"], "GET");
+ assert!(value["timestamp"].is_string());
+ }
+
+ #[test]
+ fn json_lines_without_metadata() {
+ let payload = make_json_payload(r#"{"key":"value"}"#);
+ let msg = make_message(10, payload);
+ let topic = make_topic_metadata();
+ let meta = make_messages_metadata();
+
+ let bytes =
+ format_message(&msg, &topic, &meta, false, false,
OutputFormat::JsonLines).unwrap();
+ let value: Value = serde_json::from_slice(&bytes).unwrap();
+
+ assert!(value.get("offset").is_none());
+ assert!(value.get("stream").is_none());
+ assert_eq!(value["payload"]["key"], "value");
+ }
+
+ #[test]
+ fn json_lines_with_headers() {
+ let payload = make_json_payload(r#"{"data":1}"#);
+ let mut msg = make_message(5, payload);
+
+ let mut headers = BTreeMap::new();
+ let key = iggy_common::HeaderKey::try_from("content-type").unwrap();
+ let value =
iggy_common::HeaderValue::try_from("application/json").unwrap();
+ headers.insert(key, value);
+ msg.headers = Some(headers);
+
+ let topic = make_topic_metadata();
+ let meta = make_messages_metadata();
+
+ let bytes =
+ format_message(&msg, &topic, &meta, false, true,
OutputFormat::JsonLines).unwrap();
+ let value: Value = serde_json::from_slice(&bytes).unwrap();
+
+ assert!(value["headers"].is_object());
+ assert_eq!(value["headers"]["content-type"], "application/json");
+ }
+
+ #[test]
+ fn raw_format() {
+ let payload = Payload::Text("hello world".to_string());
+ let msg = make_message(1, payload);
+ let topic = make_topic_metadata();
+ let meta = make_messages_metadata();
+
+ let bytes = format_message(&msg, &topic, &meta, true, false,
OutputFormat::Raw).unwrap();
+ assert_eq!(bytes, b"hello world");
+ }
+
+ #[test]
+ fn finalize_json_lines() {
+ let data = b"{\"a\":1}{\"b\":2}";
+ let boundaries = [7usize, 14];
+ let entries = entries_from_boundaries(data, &boundaries);
+ let result = finalize_buffer(entries, OutputFormat::JsonLines);
+ assert_eq!(result, b"{\"a\":1}\n{\"b\":2}\n");
+ }
+
+ #[test]
+ fn finalize_raw_no_delimiter() {
+ let data = b"\x00\x01\x02\x0a\xff\xfe";
+ let boundaries = [4usize, 6];
+ let entries = entries_from_boundaries(data, &boundaries);
+ let result = finalize_buffer(entries, OutputFormat::Raw);
+ assert_eq!(
+ result, data,
+ "Raw must concatenate without inserting delimiters"
+ );
+ }
+
+ #[test]
+ fn finalize_json_array() {
+ let data = b"{\"a\":1}{\"b\":2}";
+ let boundaries = [7usize, 14];
+ let entries = entries_from_boundaries(data, &boundaries);
+ let result = finalize_buffer(entries, OutputFormat::JsonArray);
+ assert_eq!(result, b"[{\"a\":1},{\"b\":2}]");
+ }
+
+ #[test]
+ fn timestamp_conversion() {
+ let ts = timestamp_to_rfc3339(1_710_597_751_000_000);
+ assert!(ts.starts_with("2024-03-16T"));
+ assert!(ts.ends_with('Z'));
+ }
+
+ #[test]
+ fn timestamp_zero() {
+ let ts = timestamp_to_rfc3339(0);
+ assert_eq!(ts, "1970-01-01T00:00:00Z");
+ }
+
+ fn entries_from_boundaries<'a>(
+ data: &'a [u8],
+ boundaries: &'a [usize],
+ ) -> impl Iterator<Item = &'a [u8]> {
+ let mut start = 0;
+ boundaries.iter().map(move |&end| {
+ let s = &data[start..end];
+ start = end;
+ s
+ })
+ }
+}
diff --git a/core/connectors/sinks/s3_sink/src/lib.rs
b/core/connectors/sinks/s3_sink/src/lib.rs
new file mode 100644
index 000000000..92a3f856c
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/lib.rs
@@ -0,0 +1,498 @@
+// 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;
+use std::str::FromStr;
+use std::sync::Arc;
+
+use iggy_connector_sdk::{Error, sink_connector};
+use secrecy::SecretString;
+use serde::{Deserialize, Serialize};
+
+mod buffer;
+mod client;
+mod formatter;
+mod path;
+mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_ATTEMPTS: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_FILE_SIZE: &str = "8MiB";
+const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}";
+const DEFAULT_OUTPUT_FORMAT: &str = "json_lines";
+const MAX_S3_SINGLE_PUT_SIZE: u64 = 5 * 1024 * 1024 * 1024; // 5 GiB
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct S3SinkConfig {
+ pub bucket: String,
+ pub region: String,
+ #[serde(default)]
+ pub prefix: Option<String>,
+ #[serde(default)]
+ pub endpoint: Option<String>,
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub access_key_id: Option<SecretString>,
+ #[serde(
+ default,
+ serialize_with = "iggy_common::serde_secret::serialize_optional_secret"
+ )]
+ pub secret_access_key: Option<SecretString>,
+ #[serde(default = "default_path_template")]
+ pub path_template: String,
+ #[serde(default = "default_file_rotation")]
+ pub file_rotation: FileRotation,
+ #[serde(default = "default_max_file_size")]
+ pub max_file_size: String,
+ #[serde(default)]
+ pub max_messages_per_file: Option<u64>,
+ #[serde(default = "default_output_format")]
+ pub output_format: String,
+ #[serde(default = "default_true")]
+ pub include_metadata: bool,
+ #[serde(default)]
+ pub include_headers: bool,
+ /// Total number of attempts (including the initial one) before giving up.
+ /// `max_attempts = 3` means 1 initial try + 2 retries. The `max_retries`
+ /// alias is accepted for convenience but follows the same "total attempts"
+ /// semantics.
+ #[serde(default, alias = "max_retries")]
+ pub max_attempts: Option<u32>,
+ #[serde(default)]
+ pub retry_delay: Option<String>,
+ #[serde(default)]
+ pub path_style: Option<bool>,
+}
+
+impl fmt::Debug for S3SinkConfig {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("S3SinkConfig")
+ .field("bucket", &self.bucket)
+ .field("region", &self.region)
+ .field("prefix", &self.prefix)
+ .field("endpoint", &self.endpoint)
+ .field("access_key_id", &"[REDACTED]")
+ .field("secret_access_key", &"[REDACTED]")
+ .field("path_template", &self.path_template)
+ .field("file_rotation", &self.file_rotation)
+ .field("max_file_size", &self.max_file_size)
+ .field("max_messages_per_file", &self.max_messages_per_file)
+ .field("output_format", &self.output_format)
+ .field("include_metadata", &self.include_metadata)
+ .field("include_headers", &self.include_headers)
+ .field("max_attempts", &self.max_attempts)
+ .field("retry_delay", &self.retry_delay)
+ .field("path_style", &self.path_style)
+ .finish()
+ }
+}
+
+fn default_path_template() -> String {
+ DEFAULT_PATH_TEMPLATE.to_string()
+}
+
+fn default_file_rotation() -> FileRotation {
+ FileRotation::Size
+}
+
+fn default_max_file_size() -> String {
+ DEFAULT_MAX_FILE_SIZE.to_string()
+}
+
+fn default_output_format() -> String {
+ DEFAULT_OUTPUT_FORMAT.to_string()
+}
+
+fn default_true() -> bool {
+ true
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FileRotation {
+ Size,
+ Messages,
+}
+
+impl fmt::Display for FileRotation {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ FileRotation::Size => write!(f, "size"),
+ FileRotation::Messages => write!(f, "messages"),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum OutputFormat {
+ JsonLines,
+ JsonArray,
+ Raw,
+}
+
+impl TryFrom<&str> for OutputFormat {
+ type Error = Error;
+
+ fn try_from(s: &str) -> Result<Self, Self::Error> {
+ match s.to_lowercase().as_str() {
+ "json_lines" | "jsonl" | "jsonlines" =>
Ok(OutputFormat::JsonLines),
+ "json_array" => Ok(OutputFormat::JsonArray),
+ "raw" => Ok(OutputFormat::Raw),
+ other => Err(Error::InvalidConfigValue(format!(
+ "Unknown output format: '{other}'. Expected: json_lines,
json_array, or raw"
+ ))),
+ }
+ }
+}
+
+impl OutputFormat {
+ pub fn file_extension(&self) -> &'static str {
+ match self {
+ OutputFormat::JsonLines => "jsonl",
+ OutputFormat::JsonArray => "json",
+ OutputFormat::Raw => "bin",
+ }
+ }
+}
+
+/// Parsed and validated config fields, constructed only inside `open()`.
+/// Avoids accessing unresolved zero/default placeholder values before
+/// the sink is fully initialized.
+#[derive(Debug)]
+pub(crate) struct ResolvedConfig {
+ pub max_file_size_bytes: u64,
+ pub max_messages: u64,
+ pub output_format: OutputFormat,
+ pub retry_delay: std::time::Duration,
+ pub max_attempts: u32,
+}
+
+pub struct S3Sink {
+ id: u32,
+ config: S3SinkConfig,
+ bucket: Option<Box<s3::Bucket>>,
+ buffers: DashMap<BufferKey, Arc<tokio::sync::Mutex<buffer::FileBuffer>>>,
+ resolved: Option<ResolvedConfig>,
+ state: tokio::sync::Mutex<SinkState>,
+}
+
+impl fmt::Debug for S3Sink {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("S3Sink")
+ .field("id", &self.id)
+ .field("config", &self.config)
+ .field("bucket", &self.bucket.as_ref().map(|b| &b.name))
+ .field("buffers_count", &self.buffers.len())
+ .field("resolved", &self.resolved)
+ .finish()
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) struct BufferKey {
+ pub stream: String,
+ pub topic: String,
+ pub partition_id: u32,
+}
+
+#[derive(Debug)]
+struct SinkState {
+ messages_received: u64,
+ messages_uploaded: u64,
+ messages_lost: u64,
+}
+
+impl S3Sink {
+ pub fn new(id: u32, config: S3SinkConfig) -> Self {
+ S3Sink {
+ id,
+ config,
+ bucket: None,
+ buffers: DashMap::new(),
+ resolved: None,
+ state: tokio::sync::Mutex::new(SinkState {
+ messages_received: 0,
+ messages_uploaded: 0,
+ messages_lost: 0,
+ }),
+ }
+ }
+
+ pub fn validate_and_parse_config(&mut self) -> Result<(), Error> {
+ if self.config.bucket.is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "bucket must not be empty".to_owned(),
+ ));
+ }
+ if self.config.region.is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "region must not be empty".to_owned(),
+ ));
+ }
+ if self.config.path_template.is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "path_template must not be empty".to_owned(),
+ ));
+ }
+
+ let output_format =
OutputFormat::try_from(self.config.output_format.as_str())?;
+ let max_file_size_bytes = parse_file_size(&self.config.max_file_size)?;
+
+ if max_file_size_bytes == 0 {
+ return Err(Error::InvalidConfigValue(
+ "max_file_size must be greater than 0".to_owned(),
+ ));
+ }
+ if max_file_size_bytes > MAX_S3_SINGLE_PUT_SIZE {
+ return Err(Error::InvalidConfigValue(format!(
+ "max_file_size ({}) exceeds S3 single PutObject limit of 5
GiB",
+ self.config.max_file_size
+ )));
+ }
+
+ let delay_str = self
+ .config
+ .retry_delay
+ .as_deref()
+ .unwrap_or(DEFAULT_RETRY_DELAY);
+ let retry_delay = humantime::Duration::from_str(delay_str)
+ .map(|d| d.into())
+ .map_err(|e| {
+ Error::InvalidConfigValue(format!("Invalid retry_delay
'{delay_str}': {e}"))
+ })?;
+
+ let mut max_messages = u64::MAX;
+ if self.config.file_rotation == FileRotation::Messages {
+ match self.config.max_messages_per_file {
+ None => {
+ return Err(Error::InvalidConfigValue(
+ "file_rotation is 'messages' but max_messages_per_file
is not configured"
+ .to_owned(),
+ ));
+ }
+ Some(0) => {
+ return Err(Error::InvalidConfigValue(
+ "max_messages_per_file must be greater than
0".to_owned(),
+ ));
+ }
+ Some(n) => {
+ max_messages = n;
+ }
+ }
+ } else if let Some(n) = self.config.max_messages_per_file {
+ if n == 0 {
+ return Err(Error::InvalidConfigValue(
+ "max_messages_per_file must be greater than 0".to_owned(),
+ ));
+ }
+ max_messages = n;
+ }
+
+ let max_attempts =
self.config.max_attempts.unwrap_or(DEFAULT_MAX_ATTEMPTS);
+
+ self.resolved = Some(ResolvedConfig {
+ max_file_size_bytes,
+ max_messages,
+ output_format,
+ retry_delay,
+ max_attempts,
+ });
+
+ Ok(())
+ }
+
+ pub(crate) fn resolved(&self) -> &ResolvedConfig {
+ self.resolved
+ .as_ref()
+ .expect("BUG: resolved config accessed before open()")
+ }
+}
+
+fn parse_file_size(s: &str) -> Result<u64, Error> {
+ byte_unit::Byte::from_str(s)
+ .map(|b| b.as_u64())
+ .map_err(|e| Error::InvalidConfigValue(format!("Invalid file size
'{s}': {e}")))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parse_file_size_mib() {
+ assert_eq!(parse_file_size("8MiB").unwrap(), 8 * 1024 * 1024);
+ }
+
+ #[test]
+ fn parse_file_size_mb() {
+ assert_eq!(parse_file_size("10MB").unwrap(), 10_000_000);
+ }
+
+ #[test]
+ fn parse_file_size_invalid() {
+ assert!(parse_file_size("not_a_size").is_err());
+ }
+
+ #[test]
+ fn output_format_json_lines_variants() {
+ assert_eq!(
+ OutputFormat::try_from("json_lines").unwrap(),
+ OutputFormat::JsonLines
+ );
+ assert_eq!(
+ OutputFormat::try_from("jsonl").unwrap(),
+ OutputFormat::JsonLines
+ );
+ assert_eq!(
+ OutputFormat::try_from("JSONLINES").unwrap(),
+ OutputFormat::JsonLines
+ );
+ }
+
+ #[test]
+ fn output_format_json_array() {
+ assert_eq!(
+ OutputFormat::try_from("json_array").unwrap(),
+ OutputFormat::JsonArray
+ );
+ }
+
+ #[test]
+ fn output_format_json_alias_removed() {
+ assert!(OutputFormat::try_from("json").is_err());
+ }
+
+ #[test]
+ fn output_format_raw() {
+ assert_eq!(OutputFormat::try_from("raw").unwrap(), OutputFormat::Raw);
+ }
+
+ #[test]
+ fn output_format_invalid() {
+ assert!(OutputFormat::try_from("xml").is_err());
+ }
+
+ #[test]
+ fn file_extensions() {
+ assert_eq!(OutputFormat::JsonLines.file_extension(), "jsonl");
+ assert_eq!(OutputFormat::JsonArray.file_extension(), "json");
+ assert_eq!(OutputFormat::Raw.file_extension(), "bin");
+ }
+
+ #[test]
+ fn file_rotation_display() {
+ assert_eq!(FileRotation::Size.to_string(), "size");
+ assert_eq!(FileRotation::Messages.to_string(), "messages");
+ }
+
+ #[test]
+ fn config_deserialization_defaults() {
+ let json = r#"{"bucket":"test","region":"us-east-1"}"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ assert_eq!(config.bucket, "test");
+ assert_eq!(config.region, "us-east-1");
+ assert_eq!(config.path_template, DEFAULT_PATH_TEMPLATE);
+ assert_eq!(config.max_file_size, DEFAULT_MAX_FILE_SIZE);
+ assert_eq!(config.output_format, DEFAULT_OUTPUT_FORMAT);
+ assert!(config.include_metadata);
+ assert!(!config.include_headers);
+ assert_eq!(config.file_rotation, FileRotation::Size);
+ assert!(config.prefix.is_none());
+ assert!(config.endpoint.is_none());
+ assert!(config.access_key_id.is_none());
+ assert!(config.secret_access_key.is_none());
+ }
+
+ #[test]
+ fn config_deserialization_full() {
+ let json = r#"{
+ "bucket": "my-bucket",
+ "region": "eu-west-1",
+ "prefix": "data/raw",
+ "endpoint": "http://localhost:9000";,
+ "access_key_id": "AKIA...",
+ "secret_access_key": "secret",
+ "path_template": "{stream}/{topic}",
+ "file_rotation": "messages",
+ "max_file_size": "16MiB",
+ "max_messages_per_file": 5000,
+ "output_format": "json_array",
+ "include_metadata": false,
+ "include_headers": true,
+ "max_attempts": 5,
+ "retry_delay": "2s",
+ "path_style": true
+ }"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ assert_eq!(config.bucket, "my-bucket");
+ assert_eq!(config.prefix.as_deref(), Some("data/raw"));
+ assert_eq!(config.endpoint.as_deref(), Some("http://localhost:9000";));
+ assert_eq!(config.file_rotation, FileRotation::Messages);
+ assert_eq!(config.max_messages_per_file, Some(5000));
+ assert!(!config.include_metadata);
+ assert!(config.include_headers);
+ assert_eq!(config.max_attempts, Some(5));
+ assert_eq!(config.path_style, Some(true));
+ }
+
+ #[test]
+ fn partial_credentials_detected() {
+ let json =
r#"{"bucket":"b","region":"us-east-1","access_key_id":"key"}"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ let result = rt.block_on(client::create_bucket(&config));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn debug_does_not_leak_secrets() {
+ let json =
+
r#"{"bucket":"b","region":"r","access_key_id":"AKIA","secret_access_key":"s3cr3t"}"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ let debug_output = format!("{:?}", config);
+ assert!(!debug_output.contains("AKIA"));
+ assert!(!debug_output.contains("s3cr3t"));
+ assert!(debug_output.contains("REDACTED"));
+ }
+
+ #[test]
+ fn s3sink_debug_does_not_leak_bucket_credentials() {
+ let json =
r#"{"bucket":"b","region":"us-east-1","access_key_id":"AKIAIOSFODNN7EXAMPLE","secret_access_key":"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ let sink = S3Sink::new(1, config);
+ let debug_output = format!("{:?}", sink);
+ assert!(
+ !debug_output.contains("AKIAIOSFODNN7EXAMPLE"),
+ "Debug output must not contain the access key"
+ );
+ assert!(
+ !debug_output.contains("wJalrXUtnFEMI"),
+ "Debug output must not contain the secret key"
+ );
+ }
+
+ #[test]
+ fn max_retries_alias_accepted() {
+ let json = r#"{"bucket":"b","region":"us-east-1","max_retries":5}"#;
+ let config: S3SinkConfig = serde_json::from_str(json).unwrap();
+ assert_eq!(config.max_attempts, Some(5));
+ }
+}
diff --git a/core/connectors/sinks/s3_sink/src/path.rs
b/core/connectors/sinks/s3_sink/src/path.rs
new file mode 100644
index 000000000..30654a524
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/path.rs
@@ -0,0 +1,245 @@
+// 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::OutputFormat;
+use chrono::{DateTime, Utc};
+use iggy_connector_sdk::Error;
+
+pub(crate) struct PathContext<'a> {
+ pub stream: &'a str,
+ pub topic: &'a str,
+ pub partition_id: u32,
+ pub first_timestamp_micros: u64,
+}
+
+pub(crate) fn render_s3_key(
+ prefix: Option<&str>,
+ template: &str,
+ ctx: &PathContext<'_>,
+ offset_start: u64,
+ offset_end: u64,
+ format: OutputFormat,
+) -> Result<String, Error> {
+ let rendered = render_template(template, ctx)?;
+
+ // Partition ID is always embedded in the filename to prevent
cross-partition
+ // key collisions (partitions have independent offset spaces starting at
0).
+ let filename = format!(
+ "{:05}-{:020}-{:020}.{}",
+ ctx.partition_id,
+ offset_start,
+ offset_end,
+ format.file_extension()
+ );
+
+ let key = match prefix {
+ Some(p) => {
+ let p = p.trim_matches('/');
+ if p.is_empty() {
+ format!("{rendered}/{filename}")
+ } else {
+ format!("{p}/{rendered}/{filename}")
+ }
+ }
+ None => format!("{rendered}/{filename}"),
+ };
+
+ Ok(key)
+}
+
+fn render_template(template: &str, ctx: &PathContext<'_>) -> Result<String,
Error> {
+ let dt = timestamp_to_datetime(ctx.first_timestamp_micros)?;
+ let date = dt.format("%Y-%m-%d").to_string();
+ let hour = dt.format("%H").to_string();
+ let ts_millis = (ctx.first_timestamp_micros / 1_000).to_string();
+
+ Ok(template
+ .replace("{stream}", &sanitize_key_segment(ctx.stream))
+ .replace("{topic}", &sanitize_key_segment(ctx.topic))
+ .replace("{partition}", &ctx.partition_id.to_string())
+ .replace("{date}", &date)
+ .replace("{hour}", &hour)
+ .replace("{timestamp}", &ts_millis))
+}
+
+/// Replace characters that produce ambiguous or hard-to-list S3 key segments.
+/// Keeps `[a-zA-Z0-9._-]`, replaces everything else with `_`.
+fn sanitize_key_segment(s: &str) -> String {
+ s.chars()
+ .map(|c| {
+ if c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-' {
+ c
+ } else {
+ '_'
+ }
+ })
+ .collect()
+}
+
+fn timestamp_to_datetime(micros: u64) -> Result<DateTime<Utc>, Error> {
+ let secs = (micros / 1_000_000) as i64;
+ let nanos = ((micros % 1_000_000) * 1_000) as u32;
+ DateTime::<Utc>::from_timestamp(secs, nanos).ok_or_else(|| {
+ Error::CannotStoreData(format!(
+ "Invalid message timestamp: {micros} micros is out of range"
+ ))
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn test_ctx() -> PathContext<'static> {
+ PathContext {
+ stream: "app_logs",
+ topic: "api_requests",
+ partition_id: 1,
+ first_timestamp_micros: 1_710_597_600_000_000, //
2024-03-16T14:00:00Z
+ }
+ }
+
+ #[test]
+ fn render_default_template() {
+ let ctx = test_ctx();
+ let key = render_s3_key(
+ Some("iggy/raw"),
+ "{stream}/{topic}/{date}/{hour}",
+ &ctx,
+ 0,
+ 99,
+ OutputFormat::JsonLines,
+ )
+ .unwrap();
+ assert_eq!(
+ key,
+
"iggy/raw/app_logs/api_requests/2024-03-16/14/00001-00000000000000000000-00000000000000000099.jsonl"
+ );
+ }
+
+ #[test]
+ fn render_with_partition() {
+ let ctx = test_ctx();
+ let key = render_s3_key(
+ None,
+ "{stream}/{topic}/{partition}/{date}",
+ &ctx,
+ 100,
+ 199,
+ OutputFormat::JsonArray,
+ )
+ .unwrap();
+ assert_eq!(
+ key,
+
"app_logs/api_requests/1/2024-03-16/00001-00000000000000000100-00000000000000000199.json"
+ );
+ }
+
+ #[test]
+ fn render_no_prefix() {
+ let ctx = test_ctx();
+ let key = render_s3_key(None, "{stream}/{topic}", &ctx, 0, 9,
OutputFormat::Raw).unwrap();
+ assert_eq!(
+ key,
+
"app_logs/api_requests/00001-00000000000000000000-00000000000000000009.bin"
+ );
+ }
+
+ #[test]
+ fn render_empty_prefix() {
+ let ctx = test_ctx();
+ let key = render_s3_key(Some(""), "{stream}", &ctx, 0, 0,
OutputFormat::JsonLines).unwrap();
+ assert_eq!(
+ key,
+ "app_logs/00001-00000000000000000000-00000000000000000000.jsonl"
+ );
+ }
+
+ #[test]
+ fn render_prefix_with_trailing_slash() {
+ let ctx = test_ctx();
+ let key = render_s3_key(
+ Some("data/"),
+ "{topic}",
+ &ctx,
+ 5,
+ 10,
+ OutputFormat::JsonLines,
+ )
+ .unwrap();
+ assert_eq!(
+ key,
+
"data/api_requests/00001-00000000000000000005-00000000000000000010.jsonl"
+ );
+ }
+
+ #[test]
+ fn timestamp_deterministic_from_message() {
+ let ctx = test_ctx();
+ let key1 = render_s3_key(None, "{timestamp}", &ctx, 0, 0,
OutputFormat::Raw).unwrap();
+ let key2 = render_s3_key(None, "{timestamp}", &ctx, 0, 0,
OutputFormat::Raw).unwrap();
+ assert_eq!(key1, key2);
+ }
+
+ #[test]
+ fn timestamp_to_datetime_zero() {
+ let dt = timestamp_to_datetime(0).unwrap();
+ assert_eq!(dt.format("%Y-%m-%d").to_string(), "1970-01-01");
+ }
+
+ #[test]
+ fn timestamp_to_datetime_known() {
+ let dt = timestamp_to_datetime(1_710_597_600_000_000).unwrap();
+ assert_eq!(dt.format("%Y-%m-%dT%H").to_string(), "2024-03-16T14");
+ }
+
+ #[test]
+ fn sanitize_stream_topic_names() {
+ let ctx = PathContext {
+ stream: "my//stream",
+ topic: "topic with spaces",
+ partition_id: 0,
+ first_timestamp_micros: 1_710_597_600_000_000,
+ };
+ let key = render_s3_key(None, "{stream}/{topic}", &ctx, 0, 0,
OutputFormat::Raw).unwrap();
+ assert!(
+ !key.contains("//"),
+ "Sanitized key must not contain '//' from stream name: {key}"
+ );
+ assert!(
+ !key.contains(' '),
+ "Sanitized key must not contain spaces: {key}"
+ );
+ }
+
+ #[test]
+ fn lex_sort_correct_with_large_offsets() {
+ let ctx = test_ctx();
+ let key_small =
+ render_s3_key(None, "{stream}", &ctx, 999_900, 999_999,
OutputFormat::Raw).unwrap();
+ let key_large = render_s3_key(
+ None,
+ "{stream}",
+ &ctx,
+ 1_000_000,
+ 1_001_000,
+ OutputFormat::Raw,
+ )
+ .unwrap();
+ assert!(key_small < key_large, "Lexicographic sort must be correct");
+ }
+}
diff --git a/core/connectors/sinks/s3_sink/src/sink.rs
b/core/connectors/sinks/s3_sink/src/sink.rs
new file mode 100644
index 000000000..59b1a59f8
--- /dev/null
+++ b/core/connectors/sinks/s3_sink/src/sink.rs
@@ -0,0 +1,662 @@
+// 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::buffer::FileBuffer;
+use crate::formatter;
+use crate::path::{PathContext, render_s3_key};
+use crate::{BufferKey, S3Sink};
+use async_trait::async_trait;
+use iggy_connector_sdk::retry::{exponential_backoff, jitter};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::{debug, error, info, warn};
+
+const MAX_BACKOFF: Duration = Duration::from_secs(60);
+
+struct FlushPayload {
+ data: Vec<u8>,
+ s3_key: String,
+ msg_count: u64,
+ first_offset: u64,
+ last_offset: u64,
+}
+
+#[async_trait]
+impl Sink for S3Sink {
+ async fn open(&mut self) -> Result<(), Error> {
+ info!("Opening S3 sink connector with ID: {}", self.id);
+
+ self.validate_and_parse_config()?;
+
+ let bucket = crate::client::create_bucket(&self.config).await?;
+
+ crate::client::verify_bucket(&bucket).await?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ self.bucket = Some(bucket);
+
+ info!(
+ "S3 sink ID: {} opened. format={}, rotation={}, max_file_size={},
template='{}'",
+ self.id,
+ self.config.output_format,
+ self.config.file_rotation,
+ self.config.max_file_size,
+ self.config.path_template,
+ );
+
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ let bucket = self
+ .bucket
+ .as_ref()
+ .ok_or_else(|| Error::InitError("S3 client not
initialized".to_string()))?;
+
+ let key = BufferKey {
+ stream: topic_metadata.stream.clone(),
+ topic: topic_metadata.topic.clone(),
+ partition_id: messages_metadata.partition_id,
+ };
+
+ let batch_size = messages.len() as u64;
+
+ {
+ let mut state = self.state.lock().await;
+ state.messages_received += batch_size;
+ }
+
+ let buffer_arc = self
+ .buffers
+ .entry(key.clone())
+ .or_insert_with(||
Arc::new(tokio::sync::Mutex::new(FileBuffer::new())))
+ .clone();
+
+ let mut processed = 0u64;
+ let result = self
+ .process_messages_inner(
+ bucket,
+ &key,
+ &buffer_arc,
+ topic_metadata,
+ &messages_metadata,
+ &messages,
+ &mut processed,
+ )
+ .await;
+
+ if let Err(ref e) = result {
+ let lost = batch_size - processed;
+ if lost > 0 {
+ let mut state = self.state.lock().await;
+ state.messages_lost += lost;
+ error!(
+ "S3 sink ID: {} lost {lost} messages from batch of
{batch_size} for {}/{}/{}: {e}",
+ self.id,
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+ }
+ }
+
+ debug!(
+ "S3 sink ID: {} buffered {} messages for {}/{}/{}",
+ self.id,
+ batch_size,
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+
+ result
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Closing S3 sink connector with ID: {}", self.id);
+
+ if let Some(bucket) = &self.bucket {
+ for entry in self.buffers.iter() {
+ let key = entry.key().clone();
+ let buffer_arc = entry.value().clone();
+ let flush_payload = {
+ let mut buffer = buffer_arc.lock().await;
+ if buffer.is_empty() {
+ None
+ } else {
+ Some(self.extract_flush_payload(&key, &mut buffer))
+ }
+ };
+ if let Some(Ok(payload)) = flush_payload {
+ if let Err(e) = self.do_upload(bucket, payload).await {
+ error!(
+ "S3 sink ID: {} failed to flush on close for
{}/{}/{}: {e}",
+ self.id, key.stream, key.topic, key.partition_id
+ );
+ }
+ } else if let Some(Err(e)) = flush_payload {
+ error!(
+ "S3 sink ID: {} failed to prepare flush on close for
{}/{}/{}: {e}",
+ self.id, key.stream, key.topic, key.partition_id
+ );
+ }
+ }
+ } else {
+ let pending: u64 = self
+ .buffers
+ .iter()
+ .map(|e| e.value().try_lock().map(|b|
b.message_count()).unwrap_or(0))
+ .sum();
+ if pending > 0 {
+ warn!(
+ "S3 sink ID: {} closing without S3 client — {pending}
buffered messages will be lost",
+ self.id,
+ );
+ }
+ }
+
+ let state = self.state.lock().await;
+ info!(
+ "S3 sink ID: {} closed. received={}, uploaded={}, lost={}",
+ self.id, state.messages_received, state.messages_uploaded,
state.messages_lost,
+ );
+
+ Ok(())
+ }
+}
+
+impl S3Sink {
+ #[allow(clippy::too_many_arguments)]
+ async fn process_messages_inner(
+ &self,
+ bucket: &s3::Bucket,
+ key: &BufferKey,
+ buffer_arc: &Arc<tokio::sync::Mutex<FileBuffer>>,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ processed: &mut u64,
+ ) -> Result<(), Error> {
+ for message in messages {
+ let resolved = self.resolved();
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ resolved.output_format,
+ )?;
+
+ let flush_payload = {
+ let mut buffer = buffer_arc.lock().await;
+ buffer.append(&formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ resolved.max_file_size_bytes,
+ resolved.max_messages,
+ ) {
+ Some(self.extract_flush_payload(key, &mut buffer)?)
+ } else {
+ None
+ }
+ };
+
+ if let Some(payload) = flush_payload {
+ self.do_upload(bucket, payload).await?;
+ }
+
+ *processed += 1;
+ }
+ Ok(())
+ }
+
+ fn extract_flush_payload(
+ &self,
+ key: &BufferKey,
+ buffer: &mut FileBuffer,
+ ) -> Result<FlushPayload, Error> {
+ let resolved = self.resolved();
+ let data = formatter::finalize_buffer(buffer.entries(),
resolved.output_format);
+
+ let ctx = PathContext {
+ stream: &key.stream,
+ topic: &key.topic,
+ partition_id: key.partition_id,
+ first_timestamp_micros: buffer.first_timestamp_micros(),
+ };
+
+ let s3_key = render_s3_key(
+ self.config.prefix.as_deref(),
+ &self.config.path_template,
+ &ctx,
+ buffer.first_offset(),
+ buffer.last_offset(),
+ resolved.output_format,
+ )?;
+
+ let msg_count = buffer.message_count();
+ let first_offset = buffer.first_offset();
+ let last_offset = buffer.last_offset();
+
+ buffer.reset();
+
+ Ok(FlushPayload {
+ data,
+ s3_key,
+ msg_count,
+ first_offset,
+ last_offset,
+ })
+ }
+
+ async fn do_upload(&self, bucket: &s3::Bucket, payload: FlushPayload) ->
Result<(), Error> {
+ match self
+ .upload_with_retry(bucket, &payload.s3_key, &payload.data)
+ .await
+ {
+ Ok(()) => {
+ debug!(
+ "S3 sink ID: {} uploaded {} ({} messages, {} bytes)",
+ self.id,
+ payload.s3_key,
+ payload.msg_count,
+ payload.data.len(),
+ );
+ let mut state = self.state.lock().await;
+ state.messages_uploaded += payload.msg_count;
+ Ok(())
+ }
+ Err(e) => {
+ error!(
+ "S3 sink ID: {} failed to upload {} ({} messages, offsets
{}-{} lost): {e}",
+ self.id,
+ payload.s3_key,
+ payload.msg_count,
+ payload.first_offset,
+ payload.last_offset,
+ );
+ let mut state = self.state.lock().await;
+ state.messages_lost += payload.msg_count;
+
+ self.write_lost_marker(
+ bucket,
+ &payload.s3_key,
+ payload.first_offset,
+ payload.last_offset,
+ payload.msg_count,
+ &e,
+ )
+ .await;
+
+ Err(e)
+ }
+ }
+ }
+
+ async fn write_lost_marker(
+ &self,
+ bucket: &s3::Bucket,
+ s3_key: &str,
+ first_offset: u64,
+ last_offset: u64,
+ msg_count: u64,
+ error: &Error,
+ ) {
+ let marker_key = format!("{s3_key}.lost");
+ let body = format!(
+ "offset_range: {first_offset}-{last_offset}\nmessage_count:
{msg_count}\nerror: {error}\n"
+ );
+ if let Err(e) = self
+ .upload_with_retry(bucket, &marker_key, body.as_bytes())
+ .await
+ {
+ warn!(
+ "S3 sink ID: {} failed to write .lost marker at {} after
retries: {e}",
+ self.id, marker_key
+ );
+ }
+ }
+
+ async fn upload_with_retry(
+ &self,
+ bucket: &s3::Bucket,
+ s3_key: &str,
+ data: &[u8],
+ ) -> Result<(), Error> {
+ let resolved = self.resolved();
+ let max_attempts = resolved.max_attempts;
+ let base_delay = resolved.retry_delay;
+ let mut attempt = 0u32;
+
+ loop {
+ match bucket.put_object(s3_key, data).await {
+ Ok(response) => {
+ let status = response.status_code();
+ if (200..300).contains(&status) {
+ return Ok(());
+ }
+
+ if !is_retriable_status(status) {
+ return Err(Error::CannotStoreData(format!(
+ "S3 PutObject returned non-retriable status
{status} for key '{s3_key}'"
+ )));
+ }
+
+ attempt += 1;
+ if attempt >= max_attempts {
+ return Err(Error::CannotStoreData(format!(
+ "S3 PutObject returned status {status} after
{max_attempts} attempts for key '{s3_key}'"
+ )));
+ }
+ warn!(
+ "S3 sink ID: {} PutObject status {status} (attempt
{attempt}/{max_attempts}). Retrying...",
+ self.id
+ );
+ }
+ Err(e) => {
+ attempt += 1;
+ if attempt >= max_attempts {
+ return Err(Error::CannotStoreData(format!(
+ "S3 PutObject failed after {max_attempts} attempts
for key '{s3_key}': {e}"
+ )));
+ }
+ warn!(
+ "S3 sink ID: {} PutObject error (attempt
{attempt}/{max_attempts}): {e}. Retrying...",
+ self.id
+ );
+ }
+ }
+ // exponential_backoff expects a 0-based retry index
+ let retry_index = attempt - 1;
+ let delay = jitter(exponential_backoff(base_delay, retry_index,
MAX_BACKOFF));
+ tokio::time::sleep(delay).await;
+ }
+ }
+}
+
+fn is_retriable_status(status: u16) -> bool {
+ status >= 500 || status == 408 || status == 429
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::{
+ DEFAULT_MAX_FILE_SIZE, DEFAULT_OUTPUT_FORMAT, DEFAULT_PATH_TEMPLATE,
FileRotation, S3Sink,
+ S3SinkConfig,
+ };
+
+ fn test_config() -> S3SinkConfig {
+ S3SinkConfig {
+ bucket: "test-bucket".to_string(),
+ region: "us-east-1".to_string(),
+ prefix: Some("data".to_string()),
+ endpoint: None,
+ access_key_id: None,
+ secret_access_key: None,
+ path_template: DEFAULT_PATH_TEMPLATE.to_string(),
+ file_rotation: FileRotation::Size,
+ max_file_size: DEFAULT_MAX_FILE_SIZE.to_string(),
+ max_messages_per_file: None,
+ output_format: DEFAULT_OUTPUT_FORMAT.to_string(),
+ include_metadata: true,
+ include_headers: false,
+ max_attempts: None,
+ retry_delay: None,
+ path_style: None,
+ }
+ }
+
+ #[test]
+ fn is_retriable_5xx() {
+ assert!(is_retriable_status(500));
+ assert!(is_retriable_status(502));
+ assert!(is_retriable_status(503));
+ assert!(is_retriable_status(599));
+ }
+
+ #[test]
+ fn is_retriable_408_429() {
+ assert!(is_retriable_status(408));
+ assert!(is_retriable_status(429));
+ }
+
+ #[test]
+ fn not_retriable_4xx() {
+ assert!(!is_retriable_status(400));
+ assert!(!is_retriable_status(403));
+ assert!(!is_retriable_status(404));
+ assert!(!is_retriable_status(405));
+ }
+
+ #[test]
+ fn not_retriable_2xx() {
+ assert!(!is_retriable_status(200));
+ assert!(!is_retriable_status(204));
+ }
+
+ #[test]
+ fn extract_flush_payload_embeds_partition_id() {
+ let config = test_config();
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+
+ let key = BufferKey {
+ stream: "logs".to_string(),
+ topic: "events".to_string(),
+ partition_id: 7,
+ };
+
+ let mut buffer = FileBuffer::new();
+ buffer.append(b"{\"a\":1}", 0, 1_710_597_600_000_000);
+ buffer.append(b"{\"b\":2}", 1, 1_710_597_601_000_000);
+
+ let payload = sink.extract_flush_payload(&key, &mut buffer).unwrap();
+ assert!(
+ payload.s3_key.contains("00007-"),
+ "S3 key must contain partition_id: {}",
+ payload.s3_key
+ );
+ assert_eq!(payload.msg_count, 2);
+ assert_eq!(payload.first_offset, 0);
+ assert_eq!(payload.last_offset, 1);
+ }
+
+ #[test]
+ fn extract_flush_payload_offset_padding_lex_sort() {
+ let config = test_config();
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+
+ let key = BufferKey {
+ stream: "s".to_string(),
+ topic: "t".to_string(),
+ partition_id: 0,
+ };
+
+ let mut buf1 = FileBuffer::new();
+ buf1.append(b"x", 999_999, 1_000_000);
+ let p1 = sink.extract_flush_payload(&key, &mut buf1).unwrap();
+
+ let mut buf2 = FileBuffer::new();
+ buf2.append(b"y", 1_000_000, 1_000_000);
+ let p2 = sink.extract_flush_payload(&key, &mut buf2).unwrap();
+
+ assert!(
+ p1.s3_key < p2.s3_key,
+ "Lex sort must be correct: {} < {}",
+ p1.s3_key,
+ p2.s3_key
+ );
+ }
+
+ #[test]
+ fn validate_max_messages_set_for_rotation_by_messages() {
+ let config = S3SinkConfig {
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: Some(500),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+ assert_eq!(sink.resolved().max_messages, 500);
+ }
+
+ #[test]
+ fn validate_rejects_messages_rotation_without_max() {
+ let config = S3SinkConfig {
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: None,
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_zero_max_messages() {
+ let config = S3SinkConfig {
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: Some(0),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_zero_max_file_size() {
+ let config = S3SinkConfig {
+ max_file_size: "0B".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_over_5gib_file_size() {
+ let config = S3SinkConfig {
+ max_file_size: "10GiB".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_empty_bucket() {
+ let config = S3SinkConfig {
+ bucket: "".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_empty_region() {
+ let config = S3SinkConfig {
+ region: "".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_rejects_empty_path_template() {
+ let config = S3SinkConfig {
+ path_template: "".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ assert!(sink.validate_and_parse_config().is_err());
+ }
+
+ #[test]
+ fn validate_size_rotation_max_messages_defaults_to_max() {
+ let config = test_config();
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+ assert_eq!(sink.resolved().max_messages, u64::MAX);
+ }
+
+ #[test]
+ fn flush_payload_data_is_json_lines() {
+ let config = test_config();
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+
+ let key = BufferKey {
+ stream: "s".to_string(),
+ topic: "t".to_string(),
+ partition_id: 0,
+ };
+
+ let mut buffer = FileBuffer::new();
+ buffer.append(b"{\"a\":1}", 0, 1_000_000);
+ buffer.append(b"{\"b\":2}", 1, 2_000_000);
+
+ let payload = sink.extract_flush_payload(&key, &mut buffer).unwrap();
+ assert_eq!(payload.data, b"{\"a\":1}\n{\"b\":2}\n");
+ }
+
+ #[test]
+ fn flush_payload_data_is_json_array() {
+ let config = S3SinkConfig {
+ output_format: "json_array".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config().unwrap();
+
+ let key = BufferKey {
+ stream: "s".to_string(),
+ topic: "t".to_string(),
+ partition_id: 0,
+ };
+
+ let mut buffer = FileBuffer::new();
+ buffer.append(b"{\"a\":1}", 0, 1_000_000);
+ buffer.append(b"{\"b\":2}", 1, 2_000_000);
+
+ let payload = sink.extract_flush_payload(&key, &mut buffer).unwrap();
+ assert_eq!(payload.data, b"[{\"a\":1},{\"b\":2}]");
+ }
+
+ #[test]
+ fn close_without_bucket_does_not_panic() {
+ let config = test_config();
+ let mut sink = S3Sink::new(1, config);
+ let rt = tokio::runtime::Runtime::new().unwrap();
+ let result = rt.block_on(sink.close());
+ assert!(result.is_ok());
+ }
+}
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index f0cddf0c0..e42d82385 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -74,6 +74,7 @@ rmcp = { workspace = true, features = [
"transport-streamable-http-client",
"transport-streamable-http-client-reqwest",
] }
+rust-s3 = { workspace = true }
secrecy = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
diff --git a/core/integration/tests/cli/common/keyring.rs
b/core/integration/tests/cli/common/keyring.rs
index 21baf946c..a6ff49543 100644
--- a/core/integration/tests/cli/common/keyring.rs
+++ b/core/integration/tests/cli/common/keyring.rs
@@ -48,4 +48,5 @@ mod backend {
pub(crate) use backend::ensure_keyring_store;
#[cfg(not(all(feature = "login-session", secret_service_keyring)))]
+#[allow(dead_code)]
pub(crate) fn ensure_keyring_store() {}
diff --git a/core/integration/tests/cli/common/mod.rs
b/core/integration/tests/cli/common/mod.rs
index 7da9f1275..b26fb1afb 100644
--- a/core/integration/tests/cli/common/mod.rs
+++ b/core/integration/tests/cli/common/mod.rs
@@ -20,6 +20,7 @@ pub(crate) mod help;
pub(crate) mod keyring;
pub(crate) use crate::cli::common::command::IggyCmdCommand;
pub(crate) use crate::cli::common::help::{CLAP_INDENT, TestHelpCmd,
USAGE_PREFIX};
+#[allow(unused_imports)]
pub(crate) use crate::cli::common::keyring::ensure_keyring_store;
use assert_cmd::assert::{Assert, OutputAssertExt};
use assert_cmd::prelude::CommandCargoExt;
diff --git a/core/integration/tests/connectors/fixtures/mod.rs
b/core/integration/tests/connectors/fixtures/mod.rs
index 0b2f264d0..369cde392 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -26,6 +26,7 @@ mod influxdb;
mod mongodb;
mod postgres;
mod quickwit;
+mod s3;
mod wiremock;
/// Prefix on every test container name so `just clean-test-containers` reaps
@@ -72,4 +73,5 @@ pub use postgres::{
PostgresSourceJsonbFixture, PostgresSourceMarkFixture, PostgresSourceOps,
};
pub use quickwit::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture};
+pub use s3::{S3SinkFixture, S3SinkOps, S3SinkRotationFixture};
pub use wiremock::{WireMockDirectFixture, WireMockWrappedFixture};
diff --git a/core/integration/tests/connectors/fixtures/s3/fixture.rs
b/core/integration/tests/connectors/fixtures/s3/fixture.rs
new file mode 100644
index 000000000..adeb03405
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/s3/fixture.rs
@@ -0,0 +1,308 @@
+// 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 integration::harness::seeds;
+use integration::harness::{TestBinaryError, TestFixture};
+use s3::creds::Credentials;
+use s3::{Bucket, Region};
+use std::collections::HashMap;
+use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy;
+use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
+use testcontainers_modules::testcontainers::runners::AsyncRunner;
+use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage,
ImageExt};
+use tracing::info;
+use uuid::Uuid;
+
+const MINIO_IMAGE: &str = "minio/minio";
+const MINIO_TAG: &str = "RELEASE.2025-09-07T16-13-09Z";
+const MINIO_PORT: u16 = 9000;
+const MINIO_CONSOLE_PORT: u16 = 9001;
+
+const MINIO_ACCESS_KEY: &str = "admin";
+const MINIO_SECRET_KEY: &str = "password";
+const MINIO_BUCKET: &str = "iggy-s3-test";
+
+const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_S3_PATH";
+const ENV_SINK_STREAMS_0_STREAM: &str =
"IGGY_CONNECTORS_SINK_S3_STREAMS_0_STREAM";
+const ENV_SINK_STREAMS_0_TOPICS: &str =
"IGGY_CONNECTORS_SINK_S3_STREAMS_0_TOPICS";
+const ENV_SINK_STREAMS_0_SCHEMA: &str =
"IGGY_CONNECTORS_SINK_S3_STREAMS_0_SCHEMA";
+const ENV_SINK_PLUGIN_BUCKET: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_BUCKET";
+const ENV_SINK_PLUGIN_REGION: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_REGION";
+const ENV_SINK_PLUGIN_ENDPOINT: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_ENDPOINT";
+const ENV_SINK_PLUGIN_PREFIX: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_PREFIX";
+const ENV_SINK_PLUGIN_ACCESS_KEY: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_ACCESS_KEY_ID";
+const ENV_SINK_PLUGIN_SECRET_KEY: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_SECRET_ACCESS_KEY";
+const ENV_SINK_PLUGIN_FILE_ROTATION: &str =
"IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_FILE_ROTATION";
+const ENV_SINK_PLUGIN_MAX_MESSAGES: &str =
+ "IGGY_CONNECTORS_SINK_S3_PLUGIN_CONFIG_MAX_MESSAGES_PER_FILE";
+
+const DEFAULT_MAX_MESSAGES_PER_FILE: usize = 5;
+const POLL_ATTEMPTS: usize = 30;
+const POLL_INTERVAL_MS: u64 = 500;
+
+pub trait S3SinkOps: Sync {
+ fn bucket(&self) -> &Bucket;
+ #[allow(dead_code)]
+ fn endpoint(&self) -> &str;
+
+ fn list_objects(
+ &self,
+ prefix: &str,
+ ) -> impl std::future::Future<Output = Result<Vec<String>,
TestBinaryError>> + Send {
+ async move {
+ let results = self
+ .bucket()
+ .list(prefix.to_string(), None)
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to list objects: {e}"),
+ })?;
+
+ let keys: Vec<String> = results
+ .iter()
+ .flat_map(|r| r.contents.iter().map(|o| o.key.clone()))
+ .collect();
+ Ok(keys)
+ }
+ }
+
+ fn get_object(
+ &self,
+ key: &str,
+ ) -> impl std::future::Future<Output = Result<Vec<u8>, TestBinaryError>> +
Send {
+ async move {
+ let response =
+ self.bucket()
+ .get_object(key)
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to get object '{key}': {e}"),
+ })?;
+ Ok(response.to_vec())
+ }
+ }
+
+ fn wait_for_objects(
+ &self,
+ prefix: &str,
+ min_objects: usize,
+ ) -> impl std::future::Future<Output = Result<Vec<String>,
TestBinaryError>> + Send {
+ async move {
+ for _ in 0..POLL_ATTEMPTS {
+ let keys = self.list_objects(prefix).await?;
+ if keys.len() >= min_objects {
+ info!(
+ "Found {} objects under prefix '{}' (required: {})",
+ keys.len(),
+ prefix,
+ min_objects
+ );
+ return Ok(keys);
+ }
+
tokio::time::sleep(std::time::Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+
+ let keys = self.list_objects(prefix).await?;
+ Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Expected at least {min_objects} objects under '{prefix}',
found {} after {POLL_ATTEMPTS} attempts",
+ keys.len()
+ ),
+ })
+ }
+ }
+}
+
+pub struct S3SinkFixture {
+ #[allow(dead_code)]
+ container: ContainerAsync<GenericImage>,
+ bucket: Box<Bucket>,
+ endpoint: String,
+}
+
+impl S3SinkOps for S3SinkFixture {
+ fn bucket(&self) -> &Bucket {
+ &self.bucket
+ }
+
+ fn endpoint(&self) -> &str {
+ &self.endpoint
+ }
+}
+
+#[async_trait]
+impl TestFixture for S3SinkFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let id = Uuid::new_v4();
+ let container_name = format!("minio-s3-{id}");
+
+ let container = GenericImage::new(MINIO_IMAGE, MINIO_TAG)
+ .with_exposed_port(MINIO_PORT.tcp())
+ .with_exposed_port(MINIO_CONSOLE_PORT.tcp())
+ .with_wait_for(WaitFor::http(
+ HttpWaitStrategy::new("/minio/health/live")
+ .with_port(MINIO_PORT.tcp())
+ .with_expected_status_code(200u16),
+ ))
+ .with_container_name(&container_name)
+ .with_env_var("MINIO_ROOT_USER", MINIO_ACCESS_KEY)
+ .with_env_var("MINIO_ROOT_PASSWORD", MINIO_SECRET_KEY)
+ .with_cmd(vec!["server", "/data", "--console-address", ":9001"])
+ .with_mapped_port(0, MINIO_PORT.tcp())
+ .with_mapped_port(0, MINIO_CONSOLE_PORT.tcp())
+ .start()
+ .await
+ .map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: format!("Failed to start MinIO container: {error}"),
+ })?;
+
+ let mapped_port = container
+ .ports()
+ .await
+ .map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: format!("Failed to get ports: {error}"),
+ })?
+ .map_to_host_port_ipv4(MINIO_PORT)
+ .ok_or_else(|| TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: "No mapping for MinIO port".to_string(),
+ })?;
+
+ let endpoint = format!("http://localhost:{mapped_port}";);
+ info!("MinIO container for S3 sink available at {endpoint}");
+
+ let region = Region::Custom {
+ region: "us-east-1".to_string(),
+ endpoint: endpoint.clone(),
+ };
+ let credentials = Credentials::new(
+ Some(MINIO_ACCESS_KEY),
+ Some(MINIO_SECRET_KEY),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: format!("Failed to create credentials: {e}"),
+ })?;
+
+ let config = s3::BucketConfiguration::default();
+ let response = Bucket::create_with_path_style(
+ MINIO_BUCKET,
+ region.clone(),
+ credentials.clone(),
+ config,
+ )
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: format!("Failed to create bucket: {e}"),
+ })?;
+ info!(
+ "S3 bucket '{}' ready (status: {})",
+ MINIO_BUCKET, response.response_code
+ );
+
+ let mut bucket = Bucket::new(MINIO_BUCKET, region,
credentials).map_err(|e| {
+ TestBinaryError::FixtureSetup {
+ fixture_type: "S3SinkFixture".to_string(),
+ message: format!("Failed to create bucket handle: {e}"),
+ }
+ })?;
+ bucket.set_path_style();
+
+ Ok(Self {
+ container,
+ bucket,
+ endpoint,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = HashMap::new();
+ envs.insert(
+ ENV_SINK_PATH.to_string(),
+ "../../target/debug/libiggy_connector_s3_sink".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_STREAM.to_string(),
+ seeds::names::STREAM.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_TOPICS.to_string(),
+ format!("[{}]", seeds::names::TOPIC),
+ );
+ envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string());
+ envs.insert(ENV_SINK_PLUGIN_BUCKET.to_string(),
MINIO_BUCKET.to_string());
+ envs.insert(ENV_SINK_PLUGIN_REGION.to_string(),
"us-east-1".to_string());
+ envs.insert(ENV_SINK_PLUGIN_ENDPOINT.to_string(),
self.endpoint.clone());
+ envs.insert(ENV_SINK_PLUGIN_PREFIX.to_string(), String::new());
+ envs.insert(
+ ENV_SINK_PLUGIN_ACCESS_KEY.to_string(),
+ MINIO_ACCESS_KEY.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_PLUGIN_SECRET_KEY.to_string(),
+ MINIO_SECRET_KEY.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_PLUGIN_FILE_ROTATION.to_string(),
+ "messages".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_PLUGIN_MAX_MESSAGES.to_string(),
+ DEFAULT_MAX_MESSAGES_PER_FILE.to_string(),
+ );
+ envs
+ }
+}
+
+pub struct S3SinkRotationFixture {
+ inner: S3SinkFixture,
+}
+
+impl S3SinkOps for S3SinkRotationFixture {
+ fn bucket(&self) -> &Bucket {
+ self.inner.bucket()
+ }
+
+ fn endpoint(&self) -> &str {
+ self.inner.endpoint()
+ }
+}
+
+#[async_trait]
+impl TestFixture for S3SinkRotationFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let inner = S3SinkFixture::setup().await?;
+ Ok(Self { inner })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.inner.connectors_runtime_envs();
+ envs.insert(
+ ENV_SINK_PLUGIN_FILE_ROTATION.to_string(),
+ "messages".to_string(),
+ );
+ envs.insert(ENV_SINK_PLUGIN_MAX_MESSAGES.to_string(),
"10".to_string());
+ envs
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/s3/mod.rs
b/core/integration/tests/connectors/fixtures/s3/mod.rs
new file mode 100644
index 000000000..fd30845a1
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/s3/mod.rs
@@ -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.
+
+mod fixture;
+
+pub use fixture::{S3SinkFixture, S3SinkOps, S3SinkRotationFixture};
diff --git a/core/integration/tests/connectors/mod.rs
b/core/integration/tests/connectors/mod.rs
index fc624f897..865712175 100644
--- a/core/integration/tests/connectors/mod.rs
+++ b/core/integration/tests/connectors/mod.rs
@@ -30,6 +30,7 @@ mod quickwit;
mod random;
mod random_source_liveness;
mod runtime;
+mod s3;
mod stdout;
use iggy_common::IggyTimestamp;
diff --git a/core/integration/tests/connectors/s3/mod.rs
b/core/integration/tests/connectors/s3/mod.rs
new file mode 100644
index 000000000..4f2135d8d
--- /dev/null
+++ b/core/integration/tests/connectors/s3/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 s3_sink;
diff --git a/core/integration/tests/connectors/s3/s3_sink.rs
b/core/integration/tests/connectors/s3/s3_sink.rs
new file mode 100644
index 000000000..22258aba2
--- /dev/null
+++ b/core/integration/tests/connectors/s3/s3_sink.rs
@@ -0,0 +1,196 @@
+// 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::create_test_messages;
+use crate::connectors::fixtures::{S3SinkFixture, S3SinkOps,
S3SinkRotationFixture};
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::Identifier;
+use iggy_common::MessageClient;
+use iggy_connector_sdk::api::SinkInfoResponse;
+use integration::harness::seeds;
+use integration::iggy_harness;
+use reqwest::Client;
+
+const API_KEY: &str = "test-api-key";
+const S3_SINK_KEY: &str = "s3";
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/s3/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn s3_sink_initializes_and_runs(harness: &TestHarness, fixture:
S3SinkFixture) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
+ let http_client = Client::new();
+
+ let response = http_client
+ .get(format!("{}/sinks", api_address))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("Failed to get sinks");
+
+ assert_eq!(response.status(), 200);
+ let sinks: Vec<SinkInfoResponse> = response.json().await.expect("Failed to
parse sinks");
+
+ assert_eq!(sinks.len(), 1);
+ assert_eq!(sinks[0].key, S3_SINK_KEY);
+ assert!(sinks[0].enabled);
+
+ drop(fixture);
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/s3/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn s3_sink_writes_jsonl_with_correct_layout(harness: &TestHarness,
fixture: S3SinkFixture) {
+ 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 message_count = 5;
+ let test_messages = create_test_messages(message_count);
+ let payloads: Vec<Bytes> = test_messages
+ .iter()
+ .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+ .collect();
+
+ let mut messages: Vec<IggyMessage> = payloads
+ .iter()
+ .enumerate()
+ .map(|(i, p)| {
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(p.clone())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ let prefix = format!("{}/", seeds::names::STREAM);
+ let keys = fixture
+ .wait_for_objects(&prefix, 1)
+ .await
+ .expect("wait for S3 objects");
+
+ assert!(!keys.is_empty(), "Expected at least one S3 object");
+
+ let key = &keys[0];
+ assert!(
+ key.contains(seeds::names::STREAM),
+ "Key must contain stream name: {key}"
+ );
+ assert!(
+ key.contains(seeds::names::TOPIC),
+ "Key must contain topic name: {key}"
+ );
+ assert!(key.ends_with(".jsonl"), "Key must end with .jsonl: {key}");
+ assert!(
+ key.contains("/00000-"),
+ "Key must contain partition_id (00000): {key}"
+ );
+
+ let data = fixture.get_object(key).await.expect("get S3 object");
+ let content = String::from_utf8(data).expect("valid utf8");
+ let lines: Vec<&str> = content.trim().lines().collect();
+ assert_eq!(
+ lines.len(),
+ message_count,
+ "Expected {message_count} lines in JSONL output"
+ );
+
+ for line in &lines {
+ let value: serde_json::Value =
serde_json::from_str(line).expect("valid JSON line");
+ assert!(value.get("offset").is_some(), "Line must have offset");
+ assert!(value.get("timestamp").is_some(), "Line must have timestamp");
+ assert!(value.get("stream").is_some(), "Line must have stream");
+ assert!(value.get("payload").is_some(), "Line must have payload");
+ }
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/s3/sink_rotation.toml")),
+ seed = seeds::connector_stream
+)]
+async fn s3_sink_rotates_on_message_count(harness: &TestHarness, fixture:
S3SinkRotationFixture) {
+ 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 message_count = 25;
+ let test_messages = create_test_messages(message_count);
+ let payloads: Vec<Bytes> = test_messages
+ .iter()
+ .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+ .collect();
+
+ let mut messages: Vec<IggyMessage> = payloads
+ .iter()
+ .enumerate()
+ .map(|(i, p)| {
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(p.clone())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ let prefix = format!("{}/", seeds::names::STREAM);
+ let keys = fixture
+ .wait_for_objects(&prefix, 2)
+ .await
+ .expect("wait for rotated S3 objects");
+
+ assert!(
+ keys.len() >= 2,
+ "Expected at least 2 S3 objects from rotation
(max_messages_per_file=10, sent 25), got {}",
+ keys.len()
+ );
+
+ for key in &keys {
+ assert!(key.ends_with(".jsonl"), "All keys must end with .jsonl");
+ assert!(
+ key.contains("/00000-"),
+ "All keys must contain partition_id"
+ );
+ }
+}
diff --git a/core/integration/tests/connectors/s3/sink.toml
b/core/integration/tests/connectors/s3/sink.toml
new file mode 100644
index 000000000..833bf6fa6
--- /dev/null
+++ b/core/integration/tests/connectors/s3/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/s3_sink"
diff --git a/core/integration/tests/connectors/s3/sink_rotation.toml
b/core/integration/tests/connectors/s3/sink_rotation.toml
new file mode 100644
index 000000000..833bf6fa6
--- /dev/null
+++ b/core/integration/tests/connectors/s3/sink_rotation.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/s3_sink"
