hubcio commented on code in PR #3103:
URL: https://github.com/apache/iggy/pull/3103#discussion_r3275880303
##########
core/connectors/sinks/s3_sink/README.md:
##########
@@ -0,0 +1,147 @@
+# 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
+- Deterministic S3 keys based on offset ranges for idempotent crash recovery
Review Comment:
"deterministic S3 keys ... idempotent crash recovery" is false on three
grounds:
- `path.rs:64` substitutes `{timestamp}` via `Utc::now().timestamp_millis()`
at render time, so retries produce different keys
- `buffer.rs` is in-memory only with no WAL, so a process restart cannot
reproduce the same batch boundaries
- runtime auto-commits offsets on poll (#2928) and the FFI return code is
discarded (#2927), so replay never happens anyway
either remove the claim, or remove `{timestamp}` from the template and
document the in-memory loss path honestly.
##########
core/connectors/sinks/s3_sink/README.md:
##########
@@ -0,0 +1,147 @@
+# 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
+- Deterministic S3 keys based on offset ranges for idempotent crash recovery
+- Optional metadata and header inclusion in output
+- Support for custom endpoints (MinIO, R2) and path-style addressing
+- Retry with exponential backoff on upload failures
Review Comment:
"retry with exponential backoff" - code at `sink.rs:260` is `retry_delay *
attempts`, which is linear (1s, 2s, 3s). either update the doc to "linear
backoff" or implement `retry_delay * 2u32.pow(attempts - 1)` with jitter. also
AFAIR there is backoff in connectors SDK, please check it.
##########
core/connectors/sinks/s3_sink/README.md:
##########
@@ -0,0 +1,147 @@
+# 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
+- Deterministic S3 keys based on offset ranges for idempotent crash recovery
+- Optional metadata and header inclusion in output
+- Support for custom endpoints (MinIO, R2) and path-style addressing
+- Retry with exponential backoff on 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_retries = 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 (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_retries` | Integer | `3` | Max upload retry attempts |
+| `retry_delay` | String | `1s` | Base delay between retries (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 | `2026-03-16` |
+| `{hour}` | UTC hour from first message | `14` |
+| `{timestamp}` | Current epoch milliseconds | `1710597600000` |
+
+### 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 = "..."
+```
+
+## Data Delivery Guarantees
+
+This connector provides **at-least-once** delivery under normal operation.
However, **data loss can occur** if all upload retries are exhausted
(controlled by `max_retries`). When an upload fails after all retry attempts,
the affected messages are dropped and an error is logged. Monitor your
connector logs for `failed to upload` errors in production. Increase
`max_retries` and `retry_delay` if transient S3 failures are common in your
environment.
Review Comment:
this paragraph claims at-least-once and then admits "data loss can occur" in
the same sentence - that is self-contradictory. given #2927 + #2928, no sink
connector can deliver at-least-once today. the canonical in-tree wording is
`http_sink/README.md:790-800` which honestly documents at-most-once and cites
both bugs. recommend copying that section verbatim.
##########
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"
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap", "once_cell", "simd-json"]
+
+[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 }
Review Comment:
`iggy_common` is declared under `[dependencies]` but the only usage in this
crate is `formatter.rs:233-235` under `#[cfg(test)]`. move it to
`[dev-dependencies]` so it does not bloat the cdylib build.
##########
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"
Review Comment:
missing `publish = false`. every other connector sink (`postgres_sink`,
`delta_sink`, `http_sink`, `elasticsearch_sink`, `stdout_sink`) declares
`publish = false` because they are cdylib plugins not meant for crates.io.
without it the crate would publish on the next workspace release.
##########
core/connectors/sinks/s3_sink/src/formatter.rs:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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, MessagesMetadata, Payload,
TopicMetadata};
+use serde_json::{Map, Value};
+
+pub fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Vec<u8> {
+ 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,
+) -> Vec<u8> {
+ let mut obj = Map::new();
+
+ if include_metadata {
+ obj.insert("offset".to_string(), Value::Number(message.offset.into()));
+ let ts = timestamp_to_rfc3339(message.timestamp);
+ obj.insert("timestamp".to_string(), Value::String(ts));
+ obj.insert(
+ "stream".to_string(),
+ Value::String(topic_metadata.stream.clone()),
+ );
+ obj.insert(
+ "topic".to_string(),
+ Value::String(topic_metadata.topic.clone()),
+ );
+ obj.insert(
+ "partition_id".to_string(),
+ Value::Number(messages_metadata.partition_id.into()),
+ );
+ }
+
+ if include_headers && let Some(headers) = &message.headers {
+ let mut headers_obj = Map::new();
+ for (key, value) in headers {
+ headers_obj.insert(key.to_string(),
Value::String(value.to_string()));
Review Comment:
`value.to_string()` invokes the `Display` impl on `HeaderField<T>` at
`core/common/src/types/message/user_headers.rs:343-348`, which writes `"{kind}:
{value}"`. emitted JSON becomes `{"content-type":"string: text/plain"}`, and
for `HeaderKind::Raw` the value at `user_headers.rs:529` is `format!("{:?}",
self.value)` - a debug-formatted `Vec<u8>` like `"raw: [104, 105]"`.
also note `HeaderKey = HeaderField<KeyMarker>` shares the same `Display`, so
`key.to_string()` produces `"string: content-type"` as the JSON map key - both
halves of the header object are mangled. downstream JSON parsers cannot recover
the original header.
fix is a `HeaderKind` match: string -> raw string, raw -> base64,
integer/float -> `Value::Number`, bool -> `Value::Bool`.
##########
core/connectors/sinks/s3_sink/src/formatter.rs:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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, MessagesMetadata, Payload,
TopicMetadata};
+use serde_json::{Map, Value};
+
+pub fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Vec<u8> {
+ 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,
+) -> Vec<u8> {
+ let mut obj = Map::new();
+
+ if include_metadata {
+ obj.insert("offset".to_string(), Value::Number(message.offset.into()));
+ let ts = timestamp_to_rfc3339(message.timestamp);
+ obj.insert("timestamp".to_string(), Value::String(ts));
+ obj.insert(
+ "stream".to_string(),
+ Value::String(topic_metadata.stream.clone()),
+ );
+ obj.insert(
+ "topic".to_string(),
+ Value::String(topic_metadata.topic.clone()),
+ );
+ obj.insert(
+ "partition_id".to_string(),
+ Value::Number(messages_metadata.partition_id.into()),
+ );
+ }
+
+ if include_headers && let Some(headers) = &message.headers {
+ let mut headers_obj = Map::new();
+ for (key, value) in headers {
+ headers_obj.insert(key.to_string(),
Value::String(value.to_string()));
+ }
+ obj.insert("headers".to_string(), Value::Object(headers_obj));
+ }
+
+ let payload_value = payload_to_json_value(&message.payload);
+ obj.insert("payload".to_string(), payload_value);
+
+ match serde_json::to_vec(&Value::Object(obj)) {
+ Ok(bytes) => bytes,
+ Err(e) => {
+ tracing::warn!(
+ "Failed to serialize message at offset {}: {e}",
+ message.offset
+ );
+ Vec::new()
+ }
+ }
+}
+
+fn format_raw_message(message: &ConsumedMessage) -> Vec<u8> {
+ message.payload.try_to_bytes().unwrap_or_default()
+}
+
+fn payload_to_json_value(payload: &Payload) -> Value {
+ match payload {
+ Payload::Json(value) => {
+ let bytes = simd_json::to_vec(value).unwrap_or_default();
+ serde_json::from_slice(&bytes).unwrap_or(Value::Null)
+ }
+ 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)),
+ }
+}
+
+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())
+}
+
+pub fn finalize_buffer(data: &[Vec<u8>], format: OutputFormat) -> Vec<u8> {
+ match format {
+ OutputFormat::JsonLines => {
+ let mut result = Vec::new();
+ for entry in data {
+ result.extend_from_slice(entry);
+ result.push(b'\n');
+ }
+ result
+ }
+ OutputFormat::JsonArray => {
+ let entries: Vec<Value> = data
+ .iter()
+ .filter_map(|bytes| serde_json::from_slice(bytes).ok())
Review Comment:
JsonArray finalize parses every already-serialized entry to `Value`,
collects to `Vec<Value>`, then `serde_json::to_vec` re-serializes the array -
triple work. also `filter_map(... .ok())` silently drops any entry that fails
to parse.
entries are already valid JSON. write `b"["`, comma-join with
`extend_from_slice`, write `b"]"`. pure byte concat, no parse, no silent drop.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_FILE_SIZE: &str = "8MiB";
+const DEFAULT_PATH_TEMPLATE: &str = "{stream}/{topic}/{date}/{hour}";
Review Comment:
the default `path_template` has no `{partition}` placeholder, and
`path.rs:40-44` hardcodes the filename as
`{first_offset:06}-{last_offset:06}.{ext}` with no partition_id either. Iggy
partition offsets are independent (both start at 0), so two partitions on the
same `stream`/`topic`/`hour` write to the identical S3 key and the last writer
wins - silent data loss on the shipped default with no misconfiguration
required.
fix needs to inject `partition_id` into the FILENAME (not just the
template), so users can't bypass it by customizing `path_template`. suggestion:
`{partition:05}-{first:020}-{last:020}.{ext}`.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
Review Comment:
`pub mod buffer`, `pub mod client`, etc. expose implementation modules as
crate-level public API for a cdylib plugin. `stdout_sink`, `postgres_sink`,
`elasticsearch_sink` all keep these helpers private. drop the `pub`.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_RETRIES: 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";
+
+#[derive(Debug, 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)]
+ pub access_key_id: Option<String>,
+ #[serde(default)]
+ pub secret_access_key: Option<String>,
+ #[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,
+ #[serde(default)]
+ pub max_retries: Option<u32>,
+ #[serde(default)]
+ pub retry_delay: Option<String>,
+ #[serde(default)]
+ pub path_style: Option<bool>,
+}
+
+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 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" | "json" => 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",
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct S3Sink {
+ id: u32,
+ config: S3SinkConfig,
+ bucket: Option<Box<s3::Bucket>>,
+ buffers: tokio::sync::Mutex<HashMap<BufferKey, buffer::FileBuffer>>,
+ max_file_size_bytes: u64,
+ output_format: OutputFormat,
+ state: tokio::sync::Mutex<SinkState>,
+ retry_delay: std::time::Duration,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct BufferKey {
+ pub stream: String,
+ pub topic: String,
+ pub partition_id: u32,
+}
+
+#[derive(Debug)]
+struct SinkState {
+ messages_processed: u64,
+ uploads_completed: u64,
+ upload_errors: u64,
+}
+
+impl S3Sink {
+ pub fn new(id: u32, config: S3SinkConfig) -> Self {
+ S3Sink {
+ id,
+ config,
+ bucket: None,
+ buffers: tokio::sync::Mutex::new(HashMap::new()),
+ max_file_size_bytes: 0,
+ output_format: OutputFormat::JsonLines,
+ state: tokio::sync::Mutex::new(SinkState {
+ messages_processed: 0,
+ uploads_completed: 0,
+ upload_errors: 0,
+ }),
+ retry_delay: std::time::Duration::from_secs(1),
+ }
+ }
+
+ pub fn validate_and_parse_config(&mut self) -> Result<(), Error> {
+ self.output_format =
OutputFormat::try_from(self.config.output_format.as_str())?;
+ self.max_file_size_bytes =
parse_file_size(&self.config.max_file_size)?;
+
+ let delay_str = self
+ .config
+ .retry_delay
+ .as_deref()
+ .unwrap_or(DEFAULT_RETRY_DELAY);
+ self.retry_delay = humantime::Duration::from_str(delay_str)
+ .map(|d| d.into())
+ .map_err(|e| {
+ Error::InvalidConfigValue(format!("Invalid retry_delay
'{delay_str}': {e}"))
+ })?;
+
+ if self.config.file_rotation == FileRotation::Messages
+ && self.config.max_messages_per_file.is_none()
+ {
+ return Err(Error::InvalidConfigValue(
+ "file_rotation is set to 'messages' but max_messages_per_file
is not configured"
+ .to_owned(),
+ ));
+ }
+
+ Ok(())
+ }
+
+ fn max_retries(&self) -> u32 {
+ self.config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES)
+ }
+}
+
+use std::str::FromStr;
Review Comment:
`use std::str::FromStr;` mid-file after the impl block. move to the top with
the other use statements.
##########
core/connectors/sinks/s3_sink/src/path.rs:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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};
+
+pub struct PathContext<'a> {
+ pub stream: &'a str,
+ pub topic: &'a str,
+ pub partition_id: u32,
+ pub first_timestamp_micros: u64,
+}
+
+pub fn render_s3_key(
+ prefix: Option<&str>,
+ template: &str,
+ ctx: &PathContext<'_>,
+ offset_start: u64,
+ offset_end: u64,
+ format: OutputFormat,
+) -> String {
+ let rendered = render_template(template, ctx);
+
+ let filename = format!(
Review Comment:
`{:06}-{:06}` zero-pads up to 6 digits but does not cap - Iggy offsets are
u64. at offset `1_000_000` the rendered name is `"1000000-1001000.jsonl"` (7
chars) and at offset `999900` it is `"999900-999999.jsonl"` (6 chars). lex sort
compares byte-by-byte and `'1' < '9'`, so the million-offset file sorts BEFORE
the 999k file in any S3 ListObjectsV2 listing.
Athena partition projection, Spark, and `aws s3 ls` all rely on lex order.
at default 1000-msg batches and modest throughput, offset 1M is a few minutes
in. pad to 20 decimal digits (max u64 = 18446744073709551615) or 16 hex digits.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
Review Comment:
`self.buffers.lock().await` holds a `tokio::sync::Mutex<HashMap<BufferKey,
FileBuffer>>` guard that lives for the entire body of `consume` - through the
for-loop at `:94-112`, through `flush_buffer` at `:110`, through
`upload_with_retry`, through `bucket.put_object().await` at `:230` and through
`tokio::time::sleep(retry_delay * attempts).await` at `:260`. one tokio task is
spawned per IggyConsumer (`runtime/sink.rs:211-216`), and all of them route to
the same `S3Sink` instance via `INSTANCES` (`sdk/sink.rs:229,241`), so this
single mutex serializes uploads across every partition of every stream/topic
this sink instance handles.
throughput ceiling = single-uploader latency. with S3 PutObject p50 50ms and
p99 200ms, that is 5-20 PUTs/sec/instance regardless of partition count.
worst-case mutex hold under failure with default retry_delay=1s + max_retries=3
is 3s blocking every partition.
notably `Cargo.toml:42` already declares `dashmap` in the dependency list
(currently unused, sitting in the `cargo-machete.ignored` set). the fix is
`DashMap<BufferKey, Arc<Mutex<FileBuffer>>>` - clone the Arc, drop the outer
guard before `flush_buffer`. `delta_sink/src/sink.rs:126-128` has the same
anti-pattern with a TODO referencing PR #2889 discussion.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ debug!(
+ "S3 sink ID: {} processed {} messages for {}/{}/{}",
+ self.id,
+ messages.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Closing S3 sink connector with ID: {}", self.id);
+
+ if let Some(bucket) = &self.bucket {
+ let mut buffers = self.buffers.lock().await;
+ let keys: Vec<BufferKey> = buffers.keys().cloned().collect();
+ for key in keys {
+ if let Some(buffer) = buffers.get_mut(&key)
+ && !buffer.is_empty()
+ {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+ } else {
+ let buffers = self.buffers.lock().await;
+ let pending: u64 = buffers.values().map(|b|
b.message_count()).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. messages_processed={},
uploads_completed={}, upload_errors={}",
+ self.id, state.messages_processed, state.uploads_completed,
state.upload_errors,
+ );
+
+ Ok(())
+ }
+}
+
+impl S3Sink {
+ async fn flush_buffer(&self, bucket: &s3::Bucket, key: &BufferKey, buffer:
&mut FileBuffer) {
+ if buffer.is_empty() {
+ return;
+ }
+
+ let data = formatter::finalize_buffer(buffer.entries(),
self.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(),
+ self.output_format,
+ );
+
+ let msg_count = buffer.message_count();
+
+ match self.upload_with_retry(bucket, &s3_key, &data).await {
+ Ok(()) => {
+ debug!(
+ "S3 sink ID: {} uploaded {} ({} messages, {} bytes)",
+ self.id,
+ s3_key,
+ msg_count,
+ data.len(),
+ );
+ let mut state = self.state.lock().await;
+ state.uploads_completed += 1;
+ drop(state);
+ buffer.reset();
+ }
+ Err(e) => {
+ error!(
+ "S3 sink ID: {} failed to upload {} ({} messages lost):
{e}",
+ self.id, s3_key, msg_count
+ );
+ let mut state = self.state.lock().await;
+ state.upload_errors += 1;
+ drop(state);
+ // Reset buffer even on failure to prevent unbounded growth.
+ // Messages are lost but offsets will be re-delivered by the
+ // runtime on next poll since consume() returned Ok.
+ buffer.reset();
Review Comment:
this is the worst data-loss path in the PR. on retry-exhaust the failure
branch logs an error and then calls `buffer.reset()` at `:214` to drop the
messages, while `consume` at `:126` still returns `Ok(())`. the comment at
`:211-213` claims "offsets will be re-delivered by the runtime on next poll
since consume() returned Ok" - that is doubly false:
- the runtime auto-commits offsets *before* consume runs
(`runtime/sink.rs:421` uses `AutoCommitWhen::PollingMessages`, see #2928), so
the offset has already advanced past these messages by the time you return.
- the runtime discards the consume FFI return code
(`runtime/sink.rs:585-593`, see #2927), so even returning `Err` wouldn't
trigger retry today.
net result: a single transient S3 hiccup that exhausts `max_retries`
(default 3) permanently loses every buffered message. the README at line 139
acknowledges this but still claims at-least-once on the same line.
minimum: drop the false comment, propagate `Err` to the runtime, and align
the README with `http_sink/README.md:790-800` (at-most-once + cite #2927 /
#2928).
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
Review Comment:
mid-batch flush failure produces an invisible offset gap in S3. trace:
append offsets 0..99 -> `flush_buffer` fails (per `:212-214` resets buffer +
logs error + returns) -> loop continues -> append offsets 100..199 -> next
flush succeeds with `000100-000199.jsonl`. there is no `000000-000099.jsonl`
and nothing in the listing tells a downstream consumer that range existed.
propagate `Err` from `flush_buffer` to `consume` so the runtime sees the
failure (which itself only gets honored once #2927 is fixed). as a stopgap,
write a `.lost` marker object describing the missing offset range.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ debug!(
+ "S3 sink ID: {} processed {} messages for {}/{}/{}",
+ self.id,
+ messages.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Closing S3 sink connector with ID: {}", self.id);
+
+ if let Some(bucket) = &self.bucket {
+ let mut buffers = self.buffers.lock().await;
+ let keys: Vec<BufferKey> = buffers.keys().cloned().collect();
+ for key in keys {
+ if let Some(buffer) = buffers.get_mut(&key)
+ && !buffer.is_empty()
+ {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+ } else {
+ let buffers = self.buffers.lock().await;
+ let pending: u64 = buffers.values().map(|b|
b.message_count()).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. messages_processed={},
uploads_completed={}, upload_errors={}",
+ self.id, state.messages_processed, state.uploads_completed,
state.upload_errors,
+ );
+
+ Ok(())
+ }
+}
+
+impl S3Sink {
+ async fn flush_buffer(&self, bucket: &s3::Bucket, key: &BufferKey, buffer:
&mut FileBuffer) {
+ if buffer.is_empty() {
+ return;
+ }
+
+ let data = formatter::finalize_buffer(buffer.entries(),
self.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(),
+ self.output_format,
+ );
+
+ let msg_count = buffer.message_count();
+
+ match self.upload_with_retry(bucket, &s3_key, &data).await {
+ Ok(()) => {
+ debug!(
+ "S3 sink ID: {} uploaded {} ({} messages, {} bytes)",
+ self.id,
+ s3_key,
+ msg_count,
+ data.len(),
+ );
+ let mut state = self.state.lock().await;
+ state.uploads_completed += 1;
+ drop(state);
+ buffer.reset();
+ }
+ Err(e) => {
+ error!(
+ "S3 sink ID: {} failed to upload {} ({} messages lost):
{e}",
+ self.id, s3_key, msg_count
+ );
+ let mut state = self.state.lock().await;
+ state.upload_errors += 1;
+ drop(state);
+ // Reset buffer even on failure to prevent unbounded growth.
+ // Messages are lost but offsets will be re-delivered by the
+ // runtime on next poll since consume() returned Ok.
+ buffer.reset();
+ }
+ }
+ }
+
+ async fn upload_with_retry(
+ &self,
+ bucket: &s3::Bucket,
+ s3_key: &str,
+ data: &[u8],
+ ) -> Result<(), Error> {
+ let max_retries = self.max_retries();
+ let retry_delay = self.retry_delay;
+ let mut attempts = 0u32;
+
+ loop {
+ match bucket.put_object(s3_key, data).await {
+ Ok(response) => {
+ let status = response.status_code();
+ if (200..300).contains(&status) {
+ return Ok(());
+ }
+ attempts += 1;
+ if attempts >= max_retries {
+ return Err(Error::CannotStoreData(format!(
+ "S3 PutObject returned status {status} after
{attempts} attempts for key '{s3_key}'"
+ )));
+ }
+ warn!(
+ "S3 sink ID: {} PutObject status {status} (attempt
{attempts}/{max_retries}). Retrying...",
+ self.id
+ );
+ }
+ Err(e) => {
+ attempts += 1;
+ if attempts >= max_retries {
+ return Err(Error::CannotStoreData(format!(
+ "S3 PutObject failed after {attempts} attempts for
key '{s3_key}': {e}"
+ )));
+ }
+ warn!(
+ "S3 sink ID: {} PutObject error (attempt
{attempts}/{max_retries}): {e}. Retrying...",
+ self.id
+ );
+ }
+ }
+ tokio::time::sleep(retry_delay * attempts).await;
Review Comment:
`retry_delay * attempts` is linear backoff (1s, 2s, 3s), but the README at
line 13 advertises exponential. either implement `retry_delay *
2u32.pow(attempts - 1)` with jitter, or update the doc to "linear".
separately, `Duration::Mul<u32>` panics on overflow - default config is safe
but pathological values (e.g. `retry_delay = "1h"` + large `max_retries`) would
panic. `saturating_mul` is cheap defensive practice.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ debug!(
+ "S3 sink ID: {} processed {} messages for {}/{}/{}",
+ self.id,
+ messages.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Closing S3 sink connector with ID: {}", self.id);
+
+ if let Some(bucket) = &self.bucket {
+ let mut buffers = self.buffers.lock().await;
+ let keys: Vec<BufferKey> = buffers.keys().cloned().collect();
+ for key in keys {
+ if let Some(buffer) = buffers.get_mut(&key)
+ && !buffer.is_empty()
+ {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+ } else {
+ let buffers = self.buffers.lock().await;
+ let pending: u64 = buffers.values().map(|b|
b.message_count()).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. messages_processed={},
uploads_completed={}, upload_errors={}",
+ self.id, state.messages_processed, state.uploads_completed,
state.upload_errors,
+ );
+
+ Ok(())
+ }
+}
+
+impl S3Sink {
+ async fn flush_buffer(&self, bucket: &s3::Bucket, key: &BufferKey, buffer:
&mut FileBuffer) {
+ if buffer.is_empty() {
+ return;
+ }
+
+ let data = formatter::finalize_buffer(buffer.entries(),
self.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(),
+ self.output_format,
+ );
+
+ let msg_count = buffer.message_count();
+
+ match self.upload_with_retry(bucket, &s3_key, &data).await {
+ Ok(()) => {
+ debug!(
+ "S3 sink ID: {} uploaded {} ({} messages, {} bytes)",
+ self.id,
+ s3_key,
+ msg_count,
+ data.len(),
+ );
+ let mut state = self.state.lock().await;
+ state.uploads_completed += 1;
+ drop(state);
+ buffer.reset();
+ }
+ Err(e) => {
+ error!(
+ "S3 sink ID: {} failed to upload {} ({} messages lost):
{e}",
+ self.id, s3_key, msg_count
+ );
+ let mut state = self.state.lock().await;
+ state.upload_errors += 1;
+ drop(state);
+ // Reset buffer even on failure to prevent unbounded growth.
+ // Messages are lost but offsets will be re-delivered by the
+ // runtime on next poll since consume() returned Ok.
+ buffer.reset();
+ }
+ }
+ }
+
+ async fn upload_with_retry(
+ &self,
+ bucket: &s3::Bucket,
+ s3_key: &str,
+ data: &[u8],
+ ) -> Result<(), Error> {
+ let max_retries = self.max_retries();
+ let retry_delay = self.retry_delay;
+ let mut attempts = 0u32;
+
+ loop {
+ match bucket.put_object(s3_key, data).await {
+ Ok(response) => {
+ let status = response.status_code();
+ if (200..300).contains(&status) {
+ return Ok(());
+ }
+ attempts += 1;
+ if attempts >= max_retries {
Review Comment:
`attempts` starts at 0 and increments before the `>= max_retries` check, so
`max_retries = 3` yields 3 total attempts (2 retries past the initial one).
this matches the postgres_sink pattern but the field name is misleading. either
rename to `max_attempts` or use `attempts > max_retries` so the field name
lines up with semantics.
##########
core/connectors/sinks/s3_sink/src/client.rs:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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 tracing::info;
+
+fn validate_credential_pair(config: &S3SinkConfig) -> Result<(), 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(),
+ ));
+ }
+ Ok(())
+}
+
+pub async fn create_bucket(config: &S3SinkConfig) -> Result<Box<Bucket>,
Error> {
+ validate_credential_pair(config)?;
+
+ let credentials = match (&config.access_key_id, &config.secret_access_key)
{
+ (Some(key), Some(secret)) => {
+ let redacted_key = key.chars().take(3).collect::<String>();
Review Comment:
`key.chars().take(3)` extracts the AWS key category prefix (`AKIA` for IAM
user, `ASIA` for STS, `AROA` for role, etc) - it reveals roughly the credential
type and zero entropy of the secret material, so it has no diagnostic value
either. log `key.len()` or the last 4 chars if you want something useful.
##########
core/connectors/sinks/s3_sink/src/client.rs:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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 tracing::info;
+
+fn validate_credential_pair(config: &S3SinkConfig) -> Result<(), 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(),
+ ));
+ }
+ Ok(())
+}
+
+pub async fn create_bucket(config: &S3SinkConfig) -> Result<Box<Bucket>,
Error> {
+ validate_credential_pair(config)?;
+
+ let credentials = match (&config.access_key_id, &config.secret_access_key)
{
+ (Some(key), Some(secret)) => {
+ let redacted_key = key.chars().take(3).collect::<String>();
+ info!("Using explicit S3 credentials (access key:
{redacted_key}***)");
+ Credentials::new(Some(key), Some(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)
+}
+
+pub async fn verify_bucket(bucket: &Bucket) -> Result<(), Error> {
+ bucket.head_object("/").await.map_err(|e| {
Review Comment:
`head_object("/")` is not a bucket connectivity check - any working bucket
returns 404 for a key named `/`, and `sink.rs:49-55` then swallows the error as
`warn!` and continues. real failures (wrong region, `SignatureDoesNotMatch`,
missing bucket, network unreachable) are all masked until the first
`put_object` inside the consume loop blows up.
rust-s3 0.37 has no `head_bucket()`. use `bucket.list_page("".into(), None,
None, None, Some(1)).await` (requires `s3:ListBucket`) or `bucket.location()`
(requires `s3:GetBucketLocation`), and treat the error as fatal in `open()`.
##########
core/connectors/sinks/s3_sink/src/formatter.rs:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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, MessagesMetadata, Payload,
TopicMetadata};
+use serde_json::{Map, Value};
+
+pub fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Vec<u8> {
+ 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,
+) -> Vec<u8> {
+ let mut obj = Map::new();
+
+ if include_metadata {
+ obj.insert("offset".to_string(), Value::Number(message.offset.into()));
+ let ts = timestamp_to_rfc3339(message.timestamp);
+ obj.insert("timestamp".to_string(), Value::String(ts));
+ obj.insert(
+ "stream".to_string(),
+ Value::String(topic_metadata.stream.clone()),
+ );
+ obj.insert(
+ "topic".to_string(),
+ Value::String(topic_metadata.topic.clone()),
+ );
+ obj.insert(
Review Comment:
`topic_metadata.stream.clone()` + `topic_metadata.topic.clone()` are
per-message clones inside `format_json_message`. for default
`batch_length=1000` that is 2000 string clones per consume call. switching this
function to a `derive(Serialize)` struct that holds `&str` references would
eliminate this plus the 5 const-key `.to_string()` allocs at `:55-69`.
##########
core/connectors/sinks/s3_sink/src/formatter.rs:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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, MessagesMetadata, Payload,
TopicMetadata};
+use serde_json::{Map, Value};
+
+pub fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Vec<u8> {
+ 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,
+) -> Vec<u8> {
+ let mut obj = Map::new();
+
+ if include_metadata {
+ obj.insert("offset".to_string(), Value::Number(message.offset.into()));
+ let ts = timestamp_to_rfc3339(message.timestamp);
+ obj.insert("timestamp".to_string(), Value::String(ts));
+ obj.insert(
+ "stream".to_string(),
+ Value::String(topic_metadata.stream.clone()),
+ );
+ obj.insert(
+ "topic".to_string(),
+ Value::String(topic_metadata.topic.clone()),
+ );
+ obj.insert(
+ "partition_id".to_string(),
+ Value::Number(messages_metadata.partition_id.into()),
+ );
+ }
+
+ if include_headers && let Some(headers) = &message.headers {
+ let mut headers_obj = Map::new();
+ for (key, value) in headers {
+ headers_obj.insert(key.to_string(),
Value::String(value.to_string()));
+ }
+ obj.insert("headers".to_string(), Value::Object(headers_obj));
+ }
+
+ let payload_value = payload_to_json_value(&message.payload);
+ obj.insert("payload".to_string(), payload_value);
+
+ match serde_json::to_vec(&Value::Object(obj)) {
+ Ok(bytes) => bytes,
+ Err(e) => {
+ tracing::warn!(
+ "Failed to serialize message at offset {}: {e}",
+ message.offset
+ );
Review Comment:
on `serde_json::to_vec` failure this returns `Vec::new()`, which
`sink.rs:103` then appends with the original offset and timestamp.
`finalize_buffer` for `JsonLines` emits an empty line for it (`{}\n` -> just
`\n`), and for `JsonArray` the `filter_map(... .ok())` at `:139` silently drops
it. either way the offset is counted as processed while the message data is
gone.
same shape at `:96` for `Raw` (`try_to_bytes().unwrap_or_default()`). this
should propagate `Result` and either skip + log + lost-counter or fail the
batch.
##########
core/connectors/sinks/s3_sink/src/buffer.rs:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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 struct FileBuffer {
+ entries: Vec<Vec<u8>>,
Review Comment:
`entries: Vec<Vec<u8>>` is one heap alloc per message on append, and
`finalize_buffer` (`formatter.rs:128-152`) then does `extend_from_slice` into a
fresh `Vec<u8>` which is a second full copy of every byte. for default
`batch_length=1000` that is 1000 small allocs plus a 2x peak-memory copy per
file.
single contiguous `Vec<u8>` + sidecar `Vec<usize>` for entry boundaries gets
this to zero extra copies on finalize and enables future `put_object_stream`
use.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_RETRIES: 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";
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
Review Comment:
`#[derive(Debug)]` on `S3SinkConfig` plus plain `Option<String>` for
`access_key_id` / `secret_access_key` means any `{:?}` log of the config (or of
`S3Sink` at `:141`, which transitively derives `Debug` via this field) prints
credentials in cleartext.
`postgres_sink/src/lib.rs:25,52` uses `secrecy::SecretString` with
`iggy_common::serde_secret::serialize_secret` for exactly this reason. this is
a regression vs the in-tree baseline.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_RETRIES: 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";
+
+#[derive(Debug, 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)]
+ pub access_key_id: Option<String>,
+ #[serde(default)]
+ pub secret_access_key: Option<String>,
+ #[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,
+ #[serde(default)]
+ pub max_retries: Option<u32>,
+ #[serde(default)]
+ pub retry_delay: Option<String>,
+ #[serde(default)]
+ pub path_style: Option<bool>,
+}
+
+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 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" | "json" => 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",
+ }
+ }
+}
+
+#[derive(Debug)]
+pub struct S3Sink {
+ id: u32,
+ config: S3SinkConfig,
+ bucket: Option<Box<s3::Bucket>>,
+ buffers: tokio::sync::Mutex<HashMap<BufferKey, buffer::FileBuffer>>,
+ max_file_size_bytes: u64,
+ output_format: OutputFormat,
+ state: tokio::sync::Mutex<SinkState>,
+ retry_delay: std::time::Duration,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct BufferKey {
+ pub stream: String,
+ pub topic: String,
+ pub partition_id: u32,
+}
+
+#[derive(Debug)]
+struct SinkState {
+ messages_processed: u64,
+ uploads_completed: u64,
+ upload_errors: u64,
+}
+
+impl S3Sink {
+ pub fn new(id: u32, config: S3SinkConfig) -> Self {
+ S3Sink {
+ id,
+ config,
+ bucket: None,
+ buffers: tokio::sync::Mutex::new(HashMap::new()),
+ max_file_size_bytes: 0,
+ output_format: OutputFormat::JsonLines,
+ state: tokio::sync::Mutex::new(SinkState {
+ messages_processed: 0,
+ uploads_completed: 0,
+ upload_errors: 0,
+ }),
+ retry_delay: std::time::Duration::from_secs(1),
+ }
+ }
+
+ pub fn validate_and_parse_config(&mut self) -> Result<(), Error> {
+ self.output_format =
OutputFormat::try_from(self.config.output_format.as_str())?;
+ self.max_file_size_bytes =
parse_file_size(&self.config.max_file_size)?;
+
+ let delay_str = self
+ .config
+ .retry_delay
+ .as_deref()
+ .unwrap_or(DEFAULT_RETRY_DELAY);
+ self.retry_delay = humantime::Duration::from_str(delay_str)
+ .map(|d| d.into())
+ .map_err(|e| {
+ Error::InvalidConfigValue(format!("Invalid retry_delay
'{delay_str}': {e}"))
+ })?;
+
+ if self.config.file_rotation == FileRotation::Messages
Review Comment:
validation gaps here cause real bugs:
- `max_messages_per_file = Some(0)` is accepted; `buffer.rs:59` then has
`count >= 0` always true after the first append, so rotation triggers on every
message -> one PutObject per message -> S3 PUT throttling at ~3500 RPS per
partitioned prefix + cost blowup.
- `max_file_size = "0B"` parses cleanly (`parse_file_size` returns `Ok(0)`);
`buffer.rs:58` has the same `>= 0` problem.
- empty `bucket` / empty `region` / empty `path_template` are all accepted.
- `max_file_size` is uncapped on the high end. rust-s3 0.37 `put_object`
does not auto-multipart, and S3 rejects single PUT > 5 GiB with
`EntityTooLarge`. a user setting `max_file_size = "10GiB"` gets opaque
retry-exhaust + silent loss per the `buffer.reset` path at `sink.rs:212-214`.
reject zero, reject empty, cap `<= 5 GiB`.
##########
core/connectors/sinks/s3_sink/src/formatter.rs:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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, MessagesMetadata, Payload,
TopicMetadata};
+use serde_json::{Map, Value};
+
+pub fn format_message(
+ message: &ConsumedMessage,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ include_metadata: bool,
+ include_headers: bool,
+ format: OutputFormat,
+) -> Vec<u8> {
+ 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,
+) -> Vec<u8> {
+ let mut obj = Map::new();
+
+ if include_metadata {
+ obj.insert("offset".to_string(), Value::Number(message.offset.into()));
+ let ts = timestamp_to_rfc3339(message.timestamp);
+ obj.insert("timestamp".to_string(), Value::String(ts));
+ obj.insert(
+ "stream".to_string(),
+ Value::String(topic_metadata.stream.clone()),
+ );
+ obj.insert(
+ "topic".to_string(),
+ Value::String(topic_metadata.topic.clone()),
+ );
+ obj.insert(
+ "partition_id".to_string(),
+ Value::Number(messages_metadata.partition_id.into()),
+ );
+ }
+
+ if include_headers && let Some(headers) = &message.headers {
+ let mut headers_obj = Map::new();
+ for (key, value) in headers {
+ headers_obj.insert(key.to_string(),
Value::String(value.to_string()));
+ }
+ obj.insert("headers".to_string(), Value::Object(headers_obj));
+ }
+
+ let payload_value = payload_to_json_value(&message.payload);
+ obj.insert("payload".to_string(), payload_value);
+
+ match serde_json::to_vec(&Value::Object(obj)) {
+ Ok(bytes) => bytes,
+ Err(e) => {
+ tracing::warn!(
+ "Failed to serialize message at offset {}: {e}",
+ message.offset
+ );
+ Vec::new()
+ }
+ }
+}
+
+fn format_raw_message(message: &ConsumedMessage) -> Vec<u8> {
+ message.payload.try_to_bytes().unwrap_or_default()
+}
+
+fn payload_to_json_value(payload: &Payload) -> Value {
+ match payload {
+ Payload::Json(value) => {
+ let bytes = simd_json::to_vec(value).unwrap_or_default();
Review Comment:
`simd_json::to_vec(value).unwrap_or_default()` ->
`serde_json::from_slice(&bytes).unwrap_or(Value::Null)` is a full round-trip
per JSON message with two silent-corruption unwraps. `sdk/src/lib.rs:50`
already re-exports `convert::owned_value_to_serde_json` which is a direct
structural converter - elasticsearch_sink, delta_sink and http_sink all use it
(`sdk/src/convert.rs:29`). one-line replacement fixes both correctness and the
perf round-trip.
##########
core/connectors/sinks/s3_sink/src/lib.rs:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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 iggy_connector_sdk::{Error, sink_connector};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::fmt;
+
+pub mod buffer;
+pub mod client;
+pub mod formatter;
+pub mod path;
+pub mod sink;
+
+sink_connector!(S3Sink);
+
+const DEFAULT_MAX_RETRIES: 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";
+
+#[derive(Debug, 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)]
+ pub access_key_id: Option<String>,
+ #[serde(default)]
+ pub secret_access_key: Option<String>,
+ #[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,
+ #[serde(default)]
+ pub max_retries: Option<u32>,
+ #[serde(default)]
+ pub retry_delay: Option<String>,
+ #[serde(default)]
+ pub path_style: Option<bool>,
+}
+
+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 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" | "json" => Ok(OutputFormat::JsonArray),
Review Comment:
`"json"` aliases to `JsonArray` here, but streaming users defaulting their
config to `output_format = "json"` will reasonably expect ndjson (`JsonLines`).
drop the ambiguous alias - keep `json_array` / `json_lines` / `jsonl` / `raw`
and reject anything else.
##########
core/connectors/sinks/s3_sink/src/path.rs:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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};
+
+pub struct PathContext<'a> {
+ pub stream: &'a str,
+ pub topic: &'a str,
+ pub partition_id: u32,
+ pub first_timestamp_micros: u64,
+}
+
+pub fn render_s3_key(
+ prefix: Option<&str>,
+ template: &str,
+ ctx: &PathContext<'_>,
+ offset_start: u64,
+ offset_end: u64,
+ format: OutputFormat,
+) -> String {
+ let rendered = render_template(template, ctx);
+
+ let filename = format!(
+ "{:06}-{:06}.{}",
+ offset_start,
+ offset_end,
+ format.file_extension()
+ );
+
+ 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}"),
+ }
+}
+
+fn render_template(template: &str, ctx: &PathContext<'_>) -> String {
+ 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 now_millis = Utc::now().timestamp_millis().to_string();
Review Comment:
`{timestamp}` substitutes `Utc::now().timestamp_millis()`, which is
wall-clock at render time and therefore non-deterministic across retries.
combined with the README idempotency claim at line 10, this is the main reason
re-uploads do not overwrite-deduplicate.
derive from `ctx.first_timestamp_micros` if a deterministic timestamp is
useful, or drop the variable entirely.
##########
core/connectors/sinks/s3_sink/src/path.rs:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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};
+
+pub struct PathContext<'a> {
+ pub stream: &'a str,
+ pub topic: &'a str,
+ pub partition_id: u32,
+ pub first_timestamp_micros: u64,
+}
+
+pub fn render_s3_key(
+ prefix: Option<&str>,
+ template: &str,
+ ctx: &PathContext<'_>,
+ offset_start: u64,
+ offset_end: u64,
+ format: OutputFormat,
+) -> String {
+ let rendered = render_template(template, ctx);
+
+ let filename = format!(
+ "{:06}-{:06}.{}",
+ offset_start,
+ offset_end,
+ format.file_extension()
+ );
+
+ 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}"),
+ }
+}
+
+fn render_template(template: &str, ctx: &PathContext<'_>) -> String {
+ 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 now_millis = Utc::now().timestamp_millis().to_string();
+
+ template
+ .replace("{stream}", ctx.stream)
+ .replace("{topic}", ctx.topic)
+ .replace("{partition}", &ctx.partition_id.to_string())
+ .replace("{date}", &date)
+ .replace("{hour}", &hour)
+ .replace("{timestamp}", &now_millis)
+}
+
+fn timestamp_to_datetime(micros: u64) -> DateTime<Utc> {
+ let secs = (micros / 1_000_000) as i64;
+ let nanos = ((micros % 1_000_000) * 1_000) as u32;
+ DateTime::<Utc>::from_timestamp(secs, nanos).unwrap_or_else(Utc::now)
Review Comment:
`from_timestamp(...).unwrap_or_else(Utc::now)` silently falls back to
wall-clock on an out-of-range Iggy timestamp. result: path partitioning
silently switches from message-time-based to ingestion-time-based without any
signal to the operator. either return an error or use `DateTime::UNIX_EPOCH`.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ debug!(
+ "S3 sink ID: {} processed {} messages for {}/{}/{}",
+ self.id,
+ messages.len(),
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ );
+
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Closing S3 sink connector with ID: {}", self.id);
+
+ if let Some(bucket) = &self.bucket {
+ let mut buffers = self.buffers.lock().await;
+ let keys: Vec<BufferKey> = buffers.keys().cloned().collect();
+ for key in keys {
+ if let Some(buffer) = buffers.get_mut(&key)
+ && !buffer.is_empty()
+ {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+ } else {
+ let buffers = self.buffers.lock().await;
+ let pending: u64 = buffers.values().map(|b|
b.message_count()).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. messages_processed={},
uploads_completed={}, upload_errors={}",
+ self.id, state.messages_processed, state.uploads_completed,
state.upload_errors,
+ );
+
+ Ok(())
+ }
+}
+
+impl S3Sink {
+ async fn flush_buffer(&self, bucket: &s3::Bucket, key: &BufferKey, buffer:
&mut FileBuffer) {
+ if buffer.is_empty() {
+ return;
+ }
+
+ let data = formatter::finalize_buffer(buffer.entries(),
self.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(),
+ self.output_format,
+ );
+
+ let msg_count = buffer.message_count();
+
+ match self.upload_with_retry(bucket, &s3_key, &data).await {
+ Ok(()) => {
+ debug!(
+ "S3 sink ID: {} uploaded {} ({} messages, {} bytes)",
+ self.id,
+ s3_key,
+ msg_count,
+ data.len(),
+ );
+ let mut state = self.state.lock().await;
+ state.uploads_completed += 1;
+ drop(state);
+ buffer.reset();
+ }
+ Err(e) => {
+ error!(
+ "S3 sink ID: {} failed to upload {} ({} messages lost):
{e}",
+ self.id, s3_key, msg_count
+ );
+ let mut state = self.state.lock().await;
+ state.upload_errors += 1;
+ drop(state);
+ // Reset buffer even on failure to prevent unbounded growth.
+ // Messages are lost but offsets will be re-delivered by the
+ // runtime on next poll since consume() returned Ok.
+ buffer.reset();
+ }
+ }
+ }
+
+ async fn upload_with_retry(
+ &self,
+ bucket: &s3::Bucket,
+ s3_key: &str,
+ data: &[u8],
+ ) -> Result<(), Error> {
+ let max_retries = self.max_retries();
+ let retry_delay = self.retry_delay;
+ let mut attempts = 0u32;
+
+ loop {
+ match bucket.put_object(s3_key, data).await {
Review Comment:
the retry loop treats every non-2xx status uniformly. AWS permanent failures
- `AccessDenied` (403), `NoSuchBucket` (404), `InvalidBucketName` (400),
`MalformedPolicy` - get retried 3 times, wasting `retry_delay * (1 + 2)` = 3s
before the final `Err`. classify retriable vs not: 5xx + 408 + 429 + 503
SlowDown -> retry; other 4xx -> fail fast.
##########
core/connectors/sinks/s3_sink/src/sink.rs:
##########
@@ -0,0 +1,263 @@
+/*
+ * 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::{ConsumedMessage, Error, MessagesMetadata, Sink,
TopicMetadata};
+use tracing::{debug, error, info, warn};
+
+#[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?;
+
+ info!(
+ "S3 sink ID: {} connected to bucket '{}' in region '{}'",
+ self.id, self.config.bucket, self.config.region
+ );
+
+ match crate::client::verify_bucket(&bucket).await {
+ Ok(()) => {
+ info!(
+ "S3 sink ID: {} bucket '{}' connectivity verified",
+ self.id, self.config.bucket
+ );
+ }
+ Err(e) => {
+ warn!(
+ "S3 sink ID: {} bucket verification returned an error
(non-fatal, \
+ the bucket may still be accessible): {e}",
+ self.id
+ );
+ }
+ }
+
+ 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 max_messages =
self.config.max_messages_per_file.unwrap_or(u64::MAX);
+
+ let mut buffers = self.buffers.lock().await;
+ let buffer =
buffers.entry(key.clone()).or_insert_with(FileBuffer::new);
+
+ for message in &messages {
+ let formatted = formatter::format_message(
+ message,
+ topic_metadata,
+ &messages_metadata,
+ self.config.include_metadata,
+ self.config.include_headers,
+ self.output_format,
+ );
+ buffer.append(formatted, message.offset, message.timestamp);
+
+ if buffer.should_rotate(
+ self.config.file_rotation,
+ self.max_file_size_bytes,
+ max_messages,
+ ) {
+ self.flush_buffer(bucket, &key, buffer).await;
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
Review Comment:
`state.messages_processed += messages.len() as u64` runs unconditionally
outside the rotate loop, so when a mid-batch flush dropped N messages this
counter still claims it processed them. `upload_errors` increments separately
at `:209`. result: the close-log at `:155-156` reports more messages processed
than actually landed in S3. either decrement on drop or split into
`messages_buffered` / `messages_uploaded` / `messages_lost`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
