You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/12 11:53:30 UTC
(iggy) branch fix/docs-audit-connector-sinks created (now b662af325)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch fix/docs-audit-connector-sinks
in repository https://gitbox.apache.org/repos/asf/iggy.git
at b662af325 fix(connectors): validate sink writes and startup readiness
This branch includes the following new commits:
new b662af325 fix(connectors): validate sink writes and startup readiness
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: fix(connectors): validate sink writes and startup readiness
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/docs-audit-connector-sinks
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b662af325fe2485988eb0628aed163f5e9807028
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Sat Sep 12 12:31:08 2026 +0200
fix(connectors): validate sink writes and startup readiness
Sink startup can accept failed probes or race backend readiness,
numeric conversion can panic or overflow, and authentication or
delivery metrics can report incorrect outcomes.
Validate ClickHouse values and PostgreSQL batch sizes, authenticate
backend checks, correct write and loss accounting, and wait for
Quickwit ingestion readiness. Keep backend regression tests and
configuration guidance with these fixes.
---
Cargo.lock | 1 +
.../example_config/connectors/quickwit_sink.toml | 2 +-
core/connectors/sinks/clickhouse_sink/README.md | 48 ++---
core/connectors/sinks/clickhouse_sink/config.toml | 13 +-
.../connectors/sinks/clickhouse_sink/src/binary.rs | 124 ++++++++++++-
core/connectors/sinks/delta_sink/README.md | 17 +-
core/connectors/sinks/delta_sink/config.toml | 2 +-
core/connectors/sinks/doris_sink/README.md | 26 +--
core/connectors/sinks/doris_sink/config.toml | 1 -
core/connectors/sinks/doris_sink/src/lib.rs | 5 +-
core/connectors/sinks/elasticsearch_sink/README.md | 4 +-
.../connectors/sinks/elasticsearch_sink/src/lib.rs | 14 +-
core/connectors/sinks/iceberg_sink/README.md | 69 ++++---
core/connectors/sinks/influxdb_sink/README.md | 89 ++++++---
core/connectors/sinks/influxdb_sink/src/lib.rs | 58 ++++--
.../connectors/sinks/influxdb_sink/src/protocol.rs | 30 +---
core/connectors/sinks/meilisearch_sink/README.md | 58 +++---
core/connectors/sinks/mongodb_sink/README.md | 74 +++++---
core/connectors/sinks/mongodb_sink/src/lib.rs | 9 -
core/connectors/sinks/postgres_sink/README.md | 107 +++++++++--
core/connectors/sinks/postgres_sink/src/lib.rs | 31 +++-
core/connectors/sinks/quickwit_sink/README.md | 22 ++-
core/connectors/sinks/quickwit_sink/config.toml | 2 +-
core/connectors/sinks/quickwit_sink/src/lib.rs | 199 ++++++++++++++++++---
core/connectors/sinks/s3_sink/Cargo.toml | 1 +
core/connectors/sinks/s3_sink/README.md | 67 ++++---
core/connectors/sinks/s3_sink/config.toml | 2 +-
core/connectors/sinks/s3_sink/src/client.rs | 42 ++++-
core/connectors/sinks/s3_sink/src/sink.rs | 116 +++++++++++-
core/connectors/sinks/surrealdb_sink/README.md | 103 ++++++++---
core/connectors/sinks/surrealdb_sink/config.toml | 2 +-
core/connectors/sinks/surrealdb_sink/src/lib.rs | 9 +-
.../connectors/elasticsearch/elasticsearch_sink.rs | 57 ++++++
core/integration/tests/connectors/fixtures/mod.rs | 4 +-
.../tests/connectors/fixtures/mongodb/sink.rs | 2 -
.../tests/connectors/fixtures/surrealdb/mod.rs | 4 +-
.../tests/connectors/fixtures/surrealdb/sink.rs | 52 +++++-
.../tests/connectors/influxdb/influxdb_sink.rs | 6 +-
.../tests/connectors/mongodb/mongodb_sink.rs | 4 +-
.../tests/connectors/surrealdb/surrealdb_sink.rs | 33 +++-
40 files changed, 1173 insertions(+), 336 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 007cf389c..790e08667 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7436,6 +7436,7 @@ dependencies = [
"simd-json",
"tokio",
"tracing",
+ "wiremock",
]
[[package]]
diff --git a/core/connectors/runtime/example_config/connectors/quickwit_sink.toml b/core/connectors/runtime/example_config/connectors/quickwit_sink.toml
index 27a3c7560..d13d02f20 100644
--- a/core/connectors/runtime/example_config/connectors/quickwit_sink.toml
+++ b/core/connectors/runtime/example_config/connectors/quickwit_sink.toml
@@ -58,7 +58,7 @@ verbose_logging = false
# max_retries = 3
# retry_delay = "1s"
# retry_max_delay = "5s"
-# Total readiness probes including the first; 1 disables retries.
+# Total attempts per readiness check including the first; 1 disables retries.
# max_open_retries = 10
# open_retry_max_delay = "30s"
# timeout = "30s"
diff --git a/core/connectors/sinks/clickhouse_sink/README.md b/core/connectors/sinks/clickhouse_sink/README.md
index 52fed58be..941d3a637 100644
--- a/core/connectors/sinks/clickhouse_sink/README.md
+++ b/core/connectors/sinks/clickhouse_sink/README.md
@@ -51,23 +51,16 @@ verbose_logging = false
| `password` | string | `""` | ClickHouse password |
| `insert_format` | string | `"json_each_row"` | Insert format: `json_each_row`, `row_binary`, or `string` |
| `string_format` | string | `"json_each_row"` | ClickHouse format for `string` mode: `json_each_row`, `csv`, or `tsv` |
-| `timeout_seconds` | u64 | `30` | HTTP request timeout |
-| `max_retries` | u32 | `3` | Max retry attempts on transient errors |
-| `retry_delay` | u64 | `1` | Delay between retries, in seconds |
+| `timeout_seconds` | u64 | `30` | HTTP request timeout in seconds |
+| `max_retries` | u32 | `3` | Total attempts for startup requests and transient insert errors; at least one even when `0` |
+| `retry_delay` | u64 | `1` | Base for exponential retry delay, in seconds |
| `verbose_logging` | bool | `false` | Log inserts at info level instead of debug |
-> **TODO:** `database` and `table` values are interpolated directly into SQL. Currently only
-> single quotes are escaped; backslashes pass through unchanged, which can misparse string
-> literals if a value ends with `\`. A future improvement should validate both fields against a
-> strict allowlist (`^[A-Za-z_][A-Za-z0-9_]*$`) at config load and escape backslashes in SQL
-> string literals. Deferred because these sinks run in operator-controlled environments where
-> config values are trusted.
-
## Insert Formats
### `json_each_row` (Default)
-Accepts messages with a `Payload::Json` payload. Each message is sent as a JSON object on its own line using ClickHouse's `JSONEachRow` format. ClickHouse handles type coercion from the JSON values to the column types, so the table can have any schema.
+Accepts messages with a `Payload::Json` payload. Each payload is serialized on its own line using ClickHouse's `JSONEachRow` format. Send JSON objects whose fields and values are compatible with the existing table and its ClickHouse input settings. The connector does not validate JSON rows against the table schema before sending them.
```toml
[plugin_config]
@@ -80,9 +73,9 @@ insert_format = "json_each_row"
Accepts messages with a `Payload::Json` payload. At startup the connector fetches the table schema from `system.columns` and validates that all column types are supported. Messages are then serialised to ClickHouse's `RowBinaryWithDefaults` binary format, which is more efficient than JSON for large volumes.
-Requires ClickHouse 23.7 or newer, when `RowBinaryWithDefaults` was introduced. Older servers reject the format; use `json_each_row` instead.
+Requires [ClickHouse 23.7 or newer](https://presentations.clickhouse.com/2023-release-23.7/index.html), when `RowBinaryWithDefaults` was introduced. Older servers reject the format; use `json_each_row` instead.
-The table must already exist. Columns with an ordinary `DEFAULT` expression can be omitted from the message — the connector emits a `0x01` prefix byte to signal that the default should be used. `MATERIALIZED`, `ALIAS`, and `EPHEMERAL` columns are not insertable and are dropped from the schema entirely.
+The table must already exist. Columns with an ordinary `DEFAULT` expression can be omitted from the message - the connector emits a `0x01` prefix byte to signal that the default should be used. An explicit JSON `null` requires a nullable column and is stored as `NULL`, even when that column has a default. Missing columns without defaults must be nullable. The connector excludes `MATERIALIZED`, `ALIAS`, and `EPHEMERAL` columns from its insert schema.
The schema is captured once at startup and never refreshed. Do not `ALTER TABLE` the target while the connector runs. See [Schema changes while running](#schema-changes-while-running).
@@ -99,7 +92,7 @@ insert_format = "row_binary"
### `string`
-Accepts messages with a `Payload::Text` payload and passes them through to ClickHouse without modification. Use `string_format` to tell ClickHouse which format to expect.
+Accepts messages with a `Payload::Text` payload and appends a newline to each payload that does not already end with one. Set the stream `schema = "text"` and use `string_format` to tell ClickHouse which format to expect.
```toml
[plugin_config]
@@ -170,17 +163,18 @@ string_format = "csv"
## Reliability
-The connector retries failed inserts up to `max_retries` times, starting from `retry_delay`. Retryable HTTP errors and network/timeout errors both back off exponentially with full jitter, so instances spread their retries instead of hammering a recovering server in lockstep. Non-retryable errors fail immediately. The startup ping and schema fetch use the same jittered backoff.
+Insert requests retry HTTP 408, 429, and 5xx responses, plus network and timeout errors. Other unsuccessful HTTP statuses fail immediately. `max_retries` is the total attempt limit, with at least one attempt even when set to `0`. Before retry number `n` (starting at 1), the delay is sampled from zero through `min(retry_delay * 2^n, 60)` seconds.
+With the defaults, there are at most three attempts and the first retry waits between zero and two seconds. The startup ping and, in `row_binary` mode, schema fetch use the same limit and backoff but retry every error.
On shutdown the connector logs the total number of messages processed.
### Bad rows in a batch
-A message whose payload type does not match the chosen format (for example a text payload in JSON mode) is always skipped with a warning. The rest of the batch is still sent.
+A message whose payload type does not match the chosen format (for example a text payload in JSON mode) is skipped with an error log. The rest of the batch is still sent.
-The `rowbinary` format has one extra case. It turns each row into binary and writes it straight into the batch buffer, so a row that cannot be converted (a value that does not fit the target column) cannot be skipped cleanly — a half-written row would corrupt the rows after it. In that case the **whole batch fails** on the first bad row and is retried as a unit per the rules above.
+The `row_binary` format fails the whole batch on the first JSON row whose values cannot be converted to the target columns. This occurs before any insert request, so the batch does not enter the plugin's insert retry loop. No partial binary row is sent.
-If a single malformed row keeps failing, every retry of that batch will fail too. Fix or remove the bad message at the source, or switch to the `json` / `string` format, which skip bad rows instead of failing the batch.
+A batch with no serializable payloads returns success without an insert. In `json_each_row` and `string` modes, ClickHouse still validates the submitted data and can reject an insert containing a malformed row.
### Schema changes while running
@@ -188,7 +182,7 @@ In `row_binary` mode the table schema is fetched once at startup and cached for
An `ALTER TABLE` that runs while the connector is live breaks that assumption. Adding, dropping, or reordering a column shifts the byte layout by one or more columns. Depending on how the shifted bytes decode, ClickHouse either rejects the batch as malformed or, worse, stores it silently with values landing in the wrong columns. Nothing detects this at runtime, so the corruption is easy to miss.
-Only `row_binary` is affected. The `json_each_row` and `string` (`json_each_row` string) formats are self-describing and map values by field name, so they tolerate schema changes.
+The `json_each_row` format, including `string` mode with `string_format = "json_each_row"`, maps values by field name, but changed column names, types, or constraints can still make inserts fail. The plain `CSV` and `TSV` string formats use the current table column order.
Until the hardening below lands, treat the `row_binary` schema as fixed for the connector's lifetime: **restart the connector after any `ALTER TABLE`** on the target table so it re-fetches the schema.
@@ -197,20 +191,16 @@ Two planned fixes remove the restriction:
1. **Explicit column list in the INSERT.** Emitting `INSERT INTO db.table (col1, col2, ...) FORMAT RowBinaryWithDefaults` binds the stream to column *names* instead of table position. ClickHouse then routes each value by name, applies `DEFAULT` for columns the connector omits, and returns a hard error (instead of silently corrupting rows) when a named column has been dropped or renamed. This makes added and reordered columns safe and turns the remaining drift into a visible failure.
2. **Refresh the schema on a failed insert.** When an insert fails with a data error, re-fetch the schema from `system.columns` and rebuild the column list before the batch is retried, letting the connector recover from a drop or rename on its own rather than failing every retry against a stale snapshot.
-### Delivery semantics: at-least-once
-
-This connector provides **at-least-once** delivery — not exactly-once. Retries resend the full batch body without an `insert_deduplication_token`, so if the server applied a batch but the acknowledgement was lost in transit (network drop, timeout), the retry will insert the same rows again.
+### Delivery semantics
-**Affected table engines:**
+The runtime uses consumer auto-commit and does not replay a failed sink batch. End-to-end at-least-once delivery is therefore not guaranteed; see [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink#sample-implementation).
-- `MergeTree` — no deduplication at all; duplicate rows will be stored.
-- `ReplicatedMergeTree` — has implicit block-level deduplication based on the data checksum (controlled by `replicated_deduplication_window`, default 100 blocks), which will suppress duplicates in the common retry case as long as the window has not been exceeded.
+Plugin retries resend the same batch without an `insert_deduplication_token`, so a lost acknowledgement can also produce duplicate rows. ClickHouse deduplication depends on the table engine, query settings, identical retry data, and the retained deduplication window:
-If your workload cannot tolerate duplicate rows, either:
+- `ReplicatedMergeTree` enables a deduplication log by default, controlled by `replicated_deduplication_window` and `replicated_deduplication_window_seconds`.
+- Non-replicated `MergeTree` can also deduplicate when `non_replicated_deduplication_window` is positive; its default is zero.
-1. Use a `ReplicatedMergeTree` table and keep `max_retries` low enough that retries stay within the deduplication window, or
-2. Use a `CollapsingMergeTree` / `ReplacingMergeTree` and apply deduplication at query time, or
-3. Accept duplicates at write time and deduplicate with `DISTINCT` or `GROUP BY` in your queries.
+See [ClickHouse insert deduplication](https://clickhouse.com/docs/concepts/features/operations/insert/deduplicating-inserts-on-retries) for the settings and limits. A small retry count alone does not ensure that a retry remains within the window, because other inserts can evict its deduplication record.
## Testing
diff --git a/core/connectors/sinks/clickhouse_sink/config.toml b/core/connectors/sinks/clickhouse_sink/config.toml
index e8eb28482..3bb217cb7 100644
--- a/core/connectors/sinks/clickhouse_sink/config.toml
+++ b/core/connectors/sinks/clickhouse_sink/config.toml
@@ -37,16 +37,13 @@ database = "default"
username = "default"
password = ""
table = "events"
-# TODO: validate database/table with a strict allowlist (^[A-Za-z_][A-Za-z0-9_]*$) and
-# escape backslashes in SQL string literals (currently only ' is escaped). Deferred
-# because sinks run in operator-controlled environments where config values are trusted.
# Insert format: "json_each_row" (default), "row_binary", or "string"
-# json_each_row — accepts Payload::Json; ClickHouse handles type coercion
-# row_binary — accepts Payload::Json; connector validates + serialises
+# json_each_row - accepts Payload::Json; ClickHouse handles type coercion
+# row_binary - accepts Payload::Json; connector validates + serialises
# to RowBinaryWithDefaults (table must exist, schema is
# fetched at startup; fails if table has unsupported types)
-# string — accepts Payload::Text; raw passthrough
+# string - accepts Payload::Text; appends a newline when absent
insert_format = "json_each_row"
# Only relevant when insert_format = "string":
@@ -54,6 +51,6 @@ insert_format = "json_each_row"
# string_format = "csv"
timeout_seconds = 30
-max_retries = 3
-retry_delay = 1 # seconds
+max_retries = 3 # total attempts, including the initial request
+retry_delay = 1 # base delay in seconds, before exponential backoff and jitter
verbose_logging = false
diff --git a/core/connectors/sinks/clickhouse_sink/src/binary.rs b/core/connectors/sinks/clickhouse_sink/src/binary.rs
index 14af11ff0..227eba0e6 100644
--- a/core/connectors/sinks/clickhouse_sink/src/binary.rs
+++ b/core/connectors/sinks/clickhouse_sink/src/binary.rs
@@ -486,7 +486,7 @@ fn coerce_i64(value: &OwnedValue) -> Result<i64, Error> {
}),
OwnedValue::Static(simd_json::StaticNode::F64(f)) => {
let n = *f as i64;
- if n as f64 != *f {
+ if !f.is_finite() || *f < i64::MIN as f64 || *f >= i64::MAX as f64 || n as f64 != *f {
error!("Float {f} is not a finite whole number within i64 range");
return Err(Error::InvalidRecord);
}
@@ -512,7 +512,7 @@ fn coerce_u64(value: &OwnedValue) -> Result<u64, Error> {
}),
OwnedValue::Static(simd_json::StaticNode::F64(f)) => {
let n = *f as u64;
- if n as f64 != *f {
+ if !f.is_finite() || *f < 0.0 || *f >= u64::MAX as f64 || n as f64 != *f {
error!("Float {f} is not a finite whole number within u64 range");
return Err(Error::InvalidRecord);
}
@@ -579,7 +579,12 @@ fn coerce_decimal(value: &OwnedValue, precision: u8, scale: u8) -> Result<i128,
to preserve full precision."
);
}
- Ok((f * 10f64.powi(i32::from(scale))).round() as i128)
+ let scaled = (f * 10f64.powi(i32::from(scale))).round();
+ if !scaled.is_finite() || scaled < i128::MIN as f64 || scaled >= i128::MAX as f64 {
+ error!("Decimal overflow for Decimal({precision}, {scale})");
+ return Err(Error::InvalidRecord);
+ }
+ Ok(scaled as i128)
}
other => {
error!("Cannot coerce {other:?} to Decimal({precision}, {scale})");
@@ -608,6 +613,15 @@ fn parse_decimal_str(s: &str, scale: u8) -> Result<i128, ()> {
None => (s, ""),
};
+ if (int_str.is_empty() && frac_str.is_empty())
+ || !int_str
+ .bytes()
+ .chain(frac_str.bytes())
+ .all(|byte| byte.is_ascii_digit())
+ {
+ return Err(());
+ }
+
let int_val: i128 = if int_str.is_empty() {
0
} else {
@@ -790,7 +804,12 @@ fn strip_timezone(s: &str) -> (&str, i64) {
match (it.next(), it.next()) {
(Some(h_s), Some(m_s)) => {
if let (Ok(h), Ok(m)) = (h_s.parse::<i64>(), m_s.parse::<i64>()) {
- Some(sign * (h * 3600 + m * 60))
+ match (h.checked_mul(3600), m.checked_mul(60)) {
+ (Some(hours), Some(minutes)) => hours
+ .checked_add(minutes)
+ .and_then(|seconds| seconds.checked_mul(sign)),
+ _ => None,
+ }
} else {
None
}
@@ -874,6 +893,65 @@ mod tests {
}
// ── primitives ───────────────────────────────────────────────────────────
+ #[test]
+ fn given_boundary_floats_when_serializing_numbers_should_preserve_values() {
+ for (column_type, value, expected) in [
+ (
+ ChType::Int64,
+ i64::MIN as f64,
+ i64::MIN.to_le_bytes().to_vec(),
+ ),
+ (
+ ChType::UInt64,
+ (1u64 << 63) as f64,
+ (1u64 << 63).to_le_bytes().to_vec(),
+ ),
+ (
+ ChType::Decimal(38, 0),
+ i128::MIN as f64,
+ i128::MIN.to_le_bytes().to_vec(),
+ ),
+ ] {
+ let mut buffer = Vec::new();
+ serialize_value(&json_f64(value), &column_type, &mut buffer).unwrap();
+ assert_eq!(buffer, expected, "value: {value}");
+ }
+ }
+
+ #[test]
+ fn given_overflowing_float_when_serializing_integer_should_reject() {
+ let mut accepted_values = Vec::new();
+ for (column_type, value) in [
+ (ChType::Int64, i64::MAX as f64),
+ (ChType::UInt64, u64::MAX as f64),
+ ] {
+ let mut row = simd_json::owned::Object::new();
+ row.insert("value".into(), json_f64(value));
+ let result = serialize_row(
+ &OwnedValue::Object(Box::new(row)),
+ &[col("value", column_type, false)],
+ &mut Vec::new(),
+ );
+ if !matches!(result, Err(Error::InvalidRecord)) {
+ accepted_values.push(value);
+ }
+ }
+ assert!(accepted_values.is_empty(), "accepted: {accepted_values:?}");
+ }
+
+ #[test]
+ fn given_overflowing_float_when_serializing_decimal_should_reject() {
+ for value in [i128::MAX as f64, f64::MAX, -f64::MAX, f64::NAN] {
+ let mut buffer = Vec::new();
+ let result = serialize_value(&json_f64(value), &ChType::Decimal(38, 0), &mut buffer);
+ assert!(
+ matches!(result, Err(Error::InvalidRecord)),
+ "value: {value}"
+ );
+ assert!(buffer.is_empty());
+ }
+ }
+
#[test]
fn serialize_int32_little_endian() {
let mut buf = vec![];
@@ -1307,6 +1385,19 @@ mod tests {
assert_eq!(buf, 1i32.to_le_bytes());
}
+ #[test]
+ fn given_overflowing_timezone_when_serializing_datetime_should_reject() {
+ for offset in [format!("{}:00", i64::MAX), format!("00:{}", i64::MAX)] {
+ let value = format!("1970-01-01T00:00:00+{offset}");
+ let result =
+ serialize_value(&json_str(&value), &ChType::DateTime64(3), &mut Vec::new());
+ assert!(
+ matches!(result, Err(Error::InvalidRecord)),
+ "value: {value}"
+ );
+ }
+ }
+
#[test]
fn serialize_datetime_from_iso8601_utc_string() {
let mut buf = vec![];
@@ -1370,6 +1461,31 @@ mod tests {
}
// ── decimal ──────────────────────────────────────────────────────────────
+ #[test]
+ fn given_multibyte_decimal_when_serializing_should_return_error() {
+ let value = format!("0.{}é", "1".repeat(37));
+ let result = serialize_value(&json_str(&value), &ChType::Decimal(38, 2), &mut Vec::new());
+ assert!(matches!(result, Err(Error::InvalidRecord)));
+ }
+
+ #[test]
+ fn given_malformed_decimal_when_serializing_should_reject() {
+ let mut accepted_values = Vec::new();
+ for value in [
+ ".".into(),
+ "--1".into(),
+ "1.-2".into(),
+ format!("0.{}x", "1".repeat(38)),
+ ] {
+ let result =
+ serialize_value(&json_str(&value), &ChType::Decimal(38, 2), &mut Vec::new());
+ if !matches!(result, Err(Error::InvalidRecord)) {
+ accepted_values.push(value);
+ }
+ }
+ assert!(accepted_values.is_empty(), "accepted: {accepted_values:?}");
+ }
+
#[test]
fn serialize_decimal32_scale2() {
let mut buf = vec![];
diff --git a/core/connectors/sinks/delta_sink/README.md b/core/connectors/sinks/delta_sink/README.md
index 84a9964c1..7a98642e4 100644
--- a/core/connectors/sinks/delta_sink/README.md
+++ b/core/connectors/sinks/delta_sink/README.md
@@ -8,6 +8,8 @@ The Delta Lake Sink Connector allows you to consume messages from Iggy topics an
- **Intelligent type coercion** to match Delta table schemas (e.g. ISO 8601 strings to timestamps)
- **Transactional writes** with atomic flush-and-commit operations
+The table must already exist. The connector appends each successful nonempty batch in one Delta transaction and keeps its schema snapshot until restart. The plugin has no failed-batch retry loop; the Delta library can retry eligible commit conflicts and storage requests. Write or commit errors clear the writer buffers and return an error. The runtime uses consumer auto-commit and does not replay failed sink batches, so end-to-end at-least-once delivery is not guaranteed.
+
## Configuration example
### Local filesystem
@@ -38,7 +40,6 @@ table_uri = "az://my-container/delta-tables/users"
storage_backend_type = "azure"
azure_storage_account_name = "mystorageaccount"
azure_storage_account_key = "account-key"
-azure_storage_sas_token = "sas-token"
azure_container_name = "my-container"
```
@@ -55,7 +56,7 @@ gcs_service_account_key = '{"type": "service_account", "project_id": "...", ...}
### Core
-- **table_uri** (required): Path or URI to the Delta table. Supported schemes: `file://`, `s3://`, `az://`, `gs://`.
+- **table_uri** (required): Absolute URI to an existing Delta table. Use `file:///...` for a local path; bare filesystem paths are not accepted. Supported schemes: `file://`, `s3://`, `az://`, `gs://`.
- **storage_backend_type** (optional): The cloud storage backend to use. One of `"s3"`, `"azure"`, or `"gcs"`. Omit for local filesystem tables.
### AWS S3
@@ -65,12 +66,12 @@ Required when `storage_backend_type = "s3"`.
- **aws_s3_access_key**: AWS access key ID.
- **aws_s3_secret_key**: AWS secret access key.
- **aws_s3_region**: AWS region (e.g. `us-east-1`).
-- **aws_s3_endpoint_url**: S3 endpoint URL. Useful for S3-compatible services like MinIO.
-- **aws_s3_allow_http**: Set to `true` to allow HTTP connections (for local development).
+- **aws_s3_endpoint_url** (optional): S3 endpoint URL. Useful for S3-compatible services like MinIO.
+- **aws_s3_allow_http** (optional, default `false`): Set to `true` to allow HTTP connections (for local development).
### Azure Blob Storage
-Required when `storage_backend_type = "azure"`.
+When `storage_backend_type = "azure"`, provide the account name, container name, and exactly one of the account key or SAS token. Providing both authentication fields is an error.
- **azure_storage_account_name**: Azure storage account name.
- **azure_storage_account_key**: Azure storage account key.
@@ -87,6 +88,6 @@ Required when `storage_backend_type = "gcs"`.
The connector automatically coerces JSON values to match the Delta table schema:
-- **Timestamp fields**: ISO 8601 / RFC 3339 formatted strings (e.g. `"2021-11-11T22:11:58Z", "2021-11-11 22:11:58"`) are converted to microsecond timestamps. Numeric timestamps pass through unchanged.
-- **String fields**: Non-string values (numbers, booleans, objects, arrays) are converted to their string representation.
-- **Nested fields**: Coercions are applied recursively to nested structs and arrays.
+- **Timestamp fields**: ISO 8601 / RFC 3339 formatted strings (e.g. `"2021-11-11T22:11:58Z", "2021-11-11 22:11:58"`) are converted to microsecond timestamps. Integer epoch-microsecond timestamps pass through unchanged. Space-separated timestamps without an offset are interpreted as UTC. Invalid timestamp strings fail the batch.
+- **String fields**: Non-null, non-string values (numbers, booleans, objects, arrays) are converted to their string representation.
+- **Nested fields**: Coercions cover nested structs, arrays of strings or timestamps, and arrays of structs. Nested arrays, maps, and variant columns pass through without these coercions. Nulls remain null.
diff --git a/core/connectors/sinks/delta_sink/config.toml b/core/connectors/sinks/delta_sink/config.toml
index 27a455593..87d7e1f8b 100644
--- a/core/connectors/sinks/delta_sink/config.toml
+++ b/core/connectors/sinks/delta_sink/config.toml
@@ -31,7 +31,7 @@ batch_length = 100
poll_interval = "5ms"
consumer_group = "delta_sink_connector"
-# table_uri: Path or URI of the Delta table to write to.
+# table_uri: Absolute URI of an existing Delta table; bare paths are not accepted.
# - Local: "file:///path/to/delta_table"
# - S3: "s3://bucket/path/to/delta_table"
# - Azure: "az://container/path/to/delta_table"
diff --git a/core/connectors/sinks/doris_sink/README.md b/core/connectors/sinks/doris_sink/README.md
index 0cea21c37..19dc9f222 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -6,8 +6,8 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
- The target Doris **database and table must be pre-created** before enabling the sink. The connector never issues DDL.
- `database` and `table` config values must match `[A-Za-z0-9_]+`. Anything else is rejected at startup with `Error::InvalidConfigValue` — this also prevents path traversal in the constructed `/api/{db}/{table}/_stream_load` URL.
-- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). If a non-JSON payload reaches the connector it logs at `error!` and aborts the whole poll; since the consumer offset is already committed at poll time the batch is not replayed — effectively silent data loss — so the upstream schema must be guaranteed JSON. (Under `schema = "json"` the SDK drops non-JSON before the connector sees it, so this abort is a defensive guard.)
-- The Iggy message JSON shape must match the target table columns. Use the optional `columns` plugin setting if the column order differs from the JSON keys.
+- Messages must arrive with `Payload::Json` (i.e. the configured stream schema is `json`). If a non-JSON payload reaches the connector it logs at `error!` and aborts the whole poll; since the consumer offset is already committed at poll time the batch is not replayed - data loss despite the error log - so the upstream schema must be guaranteed JSON. (Under `schema = "json"` the SDK drops non-JSON before the connector sees it, so this abort is a defensive guard.)
+- The Iggy message JSON shape must match the target table columns. JSON fields map by name. Use the optional `columns` plugin setting for field mappings or derived expressions.
## How it works
@@ -16,7 +16,8 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
- `hash16` is a single 64-bit blake3 hash computed over the *raw* (un-sanitized), length-prefixed `(label_prefix, table, stream, topic)` tuple.
Identities that sanitize to the same string therefore get distinct labels, whether the collision is in the names (`events.v1` vs `events_v1`) or in two tenants' prefixes that truncate alike (`prod_events_us_east_1` vs `..._2`). Length prefixes prevent boundary-shift aliasing (`("ab","c")` ≠ `("a","bc")`). The target table participates because Doris labels are scoped to a database, not a table.
- The total label is bounded under Doris's 128-char cap regardless of input length (worst case 120 chars).
- - Doris dedupes loads by label inside its `label_keep_max_second` window. The in-request retry (step 6) re-PUTs a transiently-failed batch under the same label, so a prior attempt that actually landed (e.g. a `2xx` with a missing or unreadable body) is absorbed, not doubled. This protects **in-request retry only**: the runtime commits the offset before `consume()` runs and discards its return, so a failure outliving the retry budget or a crash mid-load is **at-most-once**.
+ - Doris dedupes loads while it retains their labels (see Operational guidance). The in-request retry (step 6) re-PUTs a transiently-failed batch under the same label, so a prior attempt that actually landed (e.g. a `2xx` with a missing or unreadable body) is absorbed, not doubled.
+ The runtime uses consumer auto-commit before `consume()` finishes and does not replay a failed poll, so a failure outliving the retry budget or a crash mid-load can lose data. Expired labels or changed chunk boundaries can also permit duplicate loads during a redrive.
3. It `PUT`s the batch to `{fe_url}/api/{database}/{table}/_stream_load` with HTTP Basic auth, `Expect: 100-continue`, `label: <label>`, and the format headers (`format: json` + `strip_outer_array: true` for JSON; `format: csv` with control-char `column_separator`/`line_delimiter` + `enclose`/`escape` for CSV). `Expect: 100-continue` is required by Doris's Stream Load endpoint, which rejects PUTs that omit it; it also lets Doris reject auth/4xx before the body uploads.
4. The Doris frontend (FE) responds with a `307 Temporary Redirect` to a backend (BE). The connector follows the redirect manually so that the `Authorization` header is preserved across the hop (`reqwest`'s default policy strips it on cross-host redirects).
`308 Permanent Redirect` is also followed as a defensive measure; redirects beyond a hard cap of 5 (or a redirect with no usable `Location`) are rejected as a permanent `PermanentHttpError`, since retrying a malformed/looping redirect cannot help.
@@ -27,7 +28,8 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
- An empty or unreadable `2xx` response body → ambiguous commit outcome, retried under the same label. A non-empty malformed body remains a permanent protocol error.
- HTTP `5xx`/`408`/`429` → transient error (`Error::CannotStoreData`): retried in-request up to `max_retries` attempts (exponential backoff + jitter) under the same label before being surfaced.
- `Fail`, any other `4xx`, or a non-empty unparsable response body → permanent error (`Error::PermanentHttpError`); never retried — re-PUTing bad data would just hammer the FE.
-6. A *transient* failure (the classifications above, plus a transport-level error) is retried in-request: the same batch is re-`PUT` under the same label, up to `max_retries` attempts with backoff and ±20% jitter (`iggy_connector_sdk::retry`). Since the runtime commits the offset at poll time, this is the connector's only redelivery path; once the budget is exhausted the final attempt's error is surfaced and the batch is not retried again — **at-most-once** across polls.
+6. A *transient* failure (the classifications above, plus a transport-level error) is retried in-request: the same batch is re-`PUT` under the same label, up to `max_retries` attempts with backoff and ±20% jitter (`iggy_connector_sdk::retry`). Since the runtime commits the offset at poll time, this is the connector's only redelivery path; once the budget is exhausted the final attempt's error is surfaced and the failed poll is not automatically replayed.
+ The runtime logs and counts the error, adds no processed messages for that batch, and continues polling.
## Configuration
@@ -36,10 +38,10 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
| `fe_url` | yes | — | Doris frontend HTTP base URL, e.g. `http://localhost:8030`. |
| `database` | yes | — | Target database. Must match `[A-Za-z0-9_]+`. |
| `table` | yes | — | Target table. Must match `[A-Za-z0-9_]+`. |
-| `username` | yes | — | Doris user with `LOAD_PRIV` on the table. |
+| `username` | yes | - | Doris user authorized for Stream Load; required grants depend on the Doris version. |
| `password` | yes | — | Doris user password. Stored as a `secrecy::SecretString` and never logged. |
| `label_prefix` | no | `iggy` | Prefix for the deterministic Stream Load label. |
-| `batch_size` | no | `1000` | Maximum number of messages per Stream Load request. |
+| `batch_size` | no | `1000` | Maximum number of messages per Stream Load request; `0` is treated as `1`. |
| `timeout` | no | `30s` | Per-request HTTP timeout (total request budget), as a human-readable duration (e.g. `30s`, `1m`). |
| `connect_timeout` | no | `5s` | TCP connect timeout, independent of `timeout`, as a human-readable duration. Raise it for cross-region or cold-start FEs. |
| `max_retries` | no | `3` | Total Stream Load attempts per batch on a *transient* failure (`0` or `1` disables retries). Each retry re-PUTs under the same label, which Doris dedupes. Values above `10` are honored but emit a startup warning because they can substantially delay graceful shutdown. |
@@ -48,7 +50,7 @@ The Doris sink connector consumes JSON messages from Iggy streams and writes the
| `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream Load header. Must be a finite value in `[0.0, 1.0]`; an out-of-range value fails `open()`. |
| `columns` | no | unset | Forwarded as the `columns` Stream Load header. Validated at startup; an invalid value fails `open()`. |
| `where` | no | unset | Forwarded as the `where` Stream Load header. Validated at startup; an invalid value fails `open()`. |
-| `output_format` | no | `json` | Stream Load output format: `json` or `csv`. CSV is opt-in for throughput; it **requires `columns`** (CSV is positional, unlike name-mapped JSON) and emits control-char-framed, `enclose`/`escape`-quoted rows. `open()` fails if `output_format = "csv"` and `columns` is unset. (Named `output_format`, not `format`, to avoid an env-override collision with `plugin_config_format`.) |
+| `output_format` | no | `json` | Stream Load output format: `json` or `csv`. CSV is opt-in for throughput; it **requires `columns`** (CSV is positional, unlike name-mapped JSON) and emits control-char-framed, `enclose`/`escape`-quoted rows. `open()` fails if `output_format = "csv"` and `columns` is unset. List bare JSON field names first, followed by any derived expressions; CSV reads only the leading names before the first `=` expression. |
| `allow_insecure_redirect` | no | `false` | Permit a Stream Load redirect that downgrades `https://` → `http://`. Refused by default because it would push credentials onto a cleartext hop. |
| `allowed_redirect_hosts` | no | unset | Allowlist of redirect targets. Each entry is `host` (pins the host, any port) or `host:port` (pins the exact endpoint). When set and non-empty, any other redirect target is refused. |
@@ -61,7 +63,6 @@ enabled = true
version = 0
name = "Doris sink"
path = "target/release/libiggy_connector_doris_sink"
-plugin_config_format = "toml"
[[streams]]
stream = "events"
@@ -92,17 +93,18 @@ timeout = "30s"
## Operational guidance
-- **`label_keep_max_second`.** The connector's in-request retry re-PUTs a transiently-failed batch under the same label, so Doris must retain that label for at least the connector's full retry budget for the replay to dedupe. The Doris default is 3 days, which is conservative.
+- **Label retention.** The connector's in-request retry re-PUTs a transiently-failed batch under the same label, so Doris must retain that label for at least the connector's full retry budget for the replay to dedupe.
+ Verify the settings for your Doris version. Doris 4.0.3 uses `streaming_label_keep_max_second` (default 12 hours) for Stream Load, rather than the general `label_keep_max_second` default of three days. See [Doris FE configuration](https://doris.apache.org/docs/4.x/admin-manual/config/fe-config/#streaming_label_keep_max_second).
If you set this lower on the Doris side, use `N × 6 × timeout + (N - 1) × max_retry_delay` as a conservative per-chunk bound, where `N` is the effective total attempt count (`1` when `max_retries` is `0` or `1`), and leave operational headroom.
Each attempt can issue the initial FE request plus up to five redirected requests, and each request has its own `timeout`; every inter-attempt delay is capped at `max_retry_delay`, including jitter. Once a label expires, a retry re-loads instead of deduping, producing duplicate rows.
- **Graceful shutdown waits for in-flight retries.** Shutdown is observed between polls, not during `consume()`. An in-flight `consume()` call must return, after any remaining chunks and retries, before the plugin can close. The runtime waits up to five seconds for each consume task; if that deadline expires, the task handle is dropped and the task continues detached. The subsequent plugin close blocks while removing the SDK instance until the in-flight consume call releases its guard.
Configure `timeout`, `max_retries`, `max_retry_delay`, and `batch_size` so the per-chunk bound above, multiplied by the maximum chunks per poll (`ceil(batch_length / batch_size)`), fits your deployment's stop/restart window.
- **Label identity changed to include the target table.** Builds containing this fix generate a different hash than older builds for the same batch. This prevents two sinks targeting different tables in one database from silently deduplicating each other.
Completed batches are not replayed by an ordinary upgrade, but an old-build label never dedupes against a replay generated by a new build, even after the old request finishes. Coordinate upgrades to avoid an in-flight version boundary, and do not rely on deduplication for a cross-version manual redrive.
-- **Keep `batch_size` stable across a redrive.** The label includes the chunk's `first_offset` and `last_offset`, which are a function of `batch_size`. If you change `batch_size` between a failed load and its redrive, the chunk boundaries shift, the offsets differ, and the new label no longer matches the old one — so Doris re-loads instead of deduping, producing duplicate rows.
+- **Preserve label inputs and chunk boundaries across a redrive.** Labels include each chunk's `first_offset` and `last_offset`. Both `batch_length` and `batch_size` affect those boundaries, and actual poll sizes can vary even with unchanged settings. Different boundaries produce different labels and can reload rows instead of deduplicating them.
- **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it.
- **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load (with its own in-request retry budget for transient failures). If a chunk still fails after its retries (serialize, HTTP, or status-classification error), the connector keeps the first error, attempts the remaining chunks, and returns that error at the end — it does **not** stop at the first such failure.
- The runtime commits the consumer offset for the whole poll before `consume()` runs, so a chunk that exhausts its in-request retries is not replayed across polls; pushing the other chunks through maximizes delivered data, and the first error is surfaced at the end (logged at `error!` for observability — the runtime currently discards `consume()`'s return value, so there is no cross-poll redrive or DLQ).
+ The runtime commits the consumer offset for the whole poll before `consume()` runs, so a chunk that exhausts its in-request retries is not replayed across polls; pushing the other chunks through maximizes delivered data, and the first error is surfaced at the end (logged at `error!` for observability; the runtime counts the error and continues polling, with no automatic replay of the failed poll or DLQ).
The one deliberate exception is a **non-JSON payload**, which is treated as a schema-contract violation and aborts the whole poll immediately (see the Requirements note above). Under `schema = "json"` this is unreachable, so it is a defensive guard rather than a normal path.
## Limitations
@@ -110,4 +112,4 @@ timeout = "30s"
- Output is JSON (default) or CSV (`output_format = "csv"`, which requires `columns`). Both serialize from `Payload::Json`; raw-text/CSV passthrough and Parquet are not supported. CSV maps JSON `null` and missing keys to SQL `NULL` (`\N`), an empty string to `""`, emits numbers and booleans bare (`true`/`false`), and stringifies nested objects/arrays as JSON.
- HTTP Basic auth only.
- No automatic table creation.
-- In-request retry only. Transient backend failures are retried within a single `consume()` call (step 6), but the runtime commits the consumer offset at poll time and discards `consume()`'s return value, so there is no cross-poll redrive or DLQ — delivery is at-most-once under a crash or a failure that outlives the retry budget.
+- In-request retry only. Transient backend failures are retried within a single `consume()` call (step 6). Consumer auto-commit and the lack of automatic replay or a DLQ mean a crash or exhausted retry budget can lose data. Label retention and chunk boundaries also limit deduplication, so this is not an end-to-end exactly-once guarantee.
diff --git a/core/connectors/sinks/doris_sink/config.toml b/core/connectors/sinks/doris_sink/config.toml
index 6189e1d5e..2c595c743 100644
--- a/core/connectors/sinks/doris_sink/config.toml
+++ b/core/connectors/sinks/doris_sink/config.toml
@@ -21,7 +21,6 @@ enabled = true
version = 0
name = "Doris sink"
path = "../../target/release/libiggy_connector_doris_sink"
-plugin_config_format = "toml"
verbose = false
[[streams]]
diff --git a/core/connectors/sinks/doris_sink/src/lib.rs b/core/connectors/sinks/doris_sink/src/lib.rs
index 78e2d70d3..c28594d13 100644
--- a/core/connectors/sinks/doris_sink/src/lib.rs
+++ b/core/connectors/sinks/doris_sink/src/lib.rs
@@ -155,10 +155,7 @@ pub struct DorisSinkConfig {
pub where_clause: Option<String>,
/// Stream Load output format: `"json"` (default) or `"csv"`. CSV is opt-in
/// for throughput; because Doris CSV is positional (JSON is name-mapped), it
- /// requires `columns` to pin the column order, else `open()` fails. Named
- /// `output_format` (not `format`) so the env override
- /// `..._PLUGIN_CONFIG_OUTPUT_FORMAT` doesn't collide with the runtime's
- /// top-level `plugin_config_format` (both would flatten to `..._FORMAT`).
+ /// requires `columns` to pin the column order, else `open()` fails.
pub output_format: Option<Format>,
/// Total per-request HTTP timeout as a human-readable duration, e.g. "30s"
/// (default 30s). Matches the `timeout` field on the http/influxdb sinks.
diff --git a/core/connectors/sinks/elasticsearch_sink/README.md b/core/connectors/sinks/elasticsearch_sink/README.md
index 0c9c1038f..0c0f1cc20 100644
--- a/core/connectors/sinks/elasticsearch_sink/README.md
+++ b/core/connectors/sinks/elasticsearch_sink/README.md
@@ -7,7 +7,7 @@ A sink connector that consumes messages from Iggy streams and indexes them to El
- `url`: Elasticsearch cluster URL
- `index`: Target index name
- `username/password`: Optional authentication credentials
-- `batch_size`: Bulk indexing batch size (default: 100)
+- `batch_size`: Accepted but unused; the stream-level `batch_length` determines each incoming batch
- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed
- `create_index_if_not_exists`: Automatically create index (default: true)
- `index_mapping`: Index mapping configuration
@@ -16,6 +16,6 @@ A sink connector that consumes messages from Iggy streams and indexes them to El
- Bulk indexing optimization
- Automatic index creation
-- Error handling and retry mechanisms
+- Request errors propagate without connector retries; individual bulk-item errors are logged and counted in closing statistics but do not fail the batch
- Metadata field injection
- Support for multiple data formats
diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
index 372fd2b8f..25460cc1c 100644
--- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs
+++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs
@@ -176,9 +176,9 @@ impl ElasticsearchSink {
&self,
client: &Elasticsearch,
documents: Vec<OwnedValue>,
- ) -> Result<(), Error> {
+ ) -> Result<usize, Error> {
if documents.is_empty() {
- return Ok(());
+ return Ok(0);
}
let mut body: Vec<JsonBody<_>> = Vec::with_capacity(documents.len() * 2);
@@ -220,6 +220,7 @@ impl ElasticsearchSink {
.map_err(|e| Error::Connection(format!("Failed to parse bulk response: {}", e)))?;
// Check for individual document errors
+ let mut documents_indexed = 0;
if let Some(items) = response_body.get("items").and_then(|v| v.as_array()) {
let mut errors = 0;
for item in items {
@@ -231,12 +232,13 @@ impl ElasticsearchSink {
}
}
+ documents_indexed = items.len() - errors;
let mut state = self.state.lock().await;
state.errors_count += errors;
- state.documents_indexed += items.len() - errors;
+ state.documents_indexed += documents_indexed;
}
- Ok(())
+ Ok(documents_indexed)
}
}
@@ -350,10 +352,10 @@ impl Sink for ElasticsearchSink {
}
if !documents.is_empty() {
- self.bulk_index_documents(client, documents).await?;
+ let documents_indexed = self.bulk_index_documents(client, documents).await?;
info!(
"Successfully indexed {} documents to Elasticsearch index '{}'",
- messages_count, self.config.index
+ documents_indexed, self.config.index
);
}
diff --git a/core/connectors/sinks/iceberg_sink/README.md b/core/connectors/sinks/iceberg_sink/README.md
index 386a6d417..a2f3ec4d3 100644
--- a/core/connectors/sinks/iceberg_sink/README.md
+++ b/core/connectors/sinks/iceberg_sink/README.md
@@ -12,13 +12,17 @@ The Iceberg Sink Connector allows you to consume messages from Iggy topics and s
## Configuration example
+The catalog, storage bucket, namespace and target tables must already exist. Save this
+`plugin_config` section in a complete sink connector file with `schema = "json"` on the stream.
+The sink does not create tables.
+
```toml
[plugin_config]
tables = ["nyc.users"]
catalog_type = "rest"
warehouse = "warehouse"
uri = "http://localhost:8181"
-dynamic_routing = true
+dynamic_routing = false
dynamic_route_field = "db_table"
store_url = "http://localhost:9000"
store_access_key_id = "admin"
@@ -30,34 +34,49 @@ store_path_style_access = true
## Configuration Options
-- **tables**: The names of the Iceberg tables you want to statically route Iggy messages to. The name should include the table’s namespace, separated by a dot (`.`).
+- **tables**: The names of the Iceberg tables you want to statically route Iggy messages to. The name should include the table’s namespace, separated by a dot (`.`). Ignored when `dynamic_routing = true`.
- **catalog_type**: The type of catalog you are routing data to. **Currently, only REST catalogs are fully supported.**
-- **warehouse**: The name of the bucket or warehouse where Iggy will upload data files.
-- **URI**: The URI of the Iceberg catalog.
+- **warehouse**: The warehouse value sent to the REST catalog. Its meaning depends on the catalog service; data file destinations come from each table’s metadata.
+- **uri**: The URI of the Iceberg catalog.
- **dynamic_routing**: Enables dynamic routing. See more details later in this document.
-- **dynamic_route_field**: The name of the message field that specifies the Iceberg table to route data to. See more details below.
+- **dynamic_route_field**: The top-level message field that specifies the Iceberg table to route data to. Ignored in static mode. See more details below.
- **store_url**: The URL of the object storage for data uploads.
-- **store_access_key_id**: The access key ID of the object storage.
-- **store_secret_access_key**: The secret key used to upload data to the object storage.
-- **store_region**: The region of the object storage, if applicable.
+- **store_access_key_id**: The optional access key ID of the object storage.
+- **store_secret_access_key**: The optional secret key used to upload data to the object storage. Supply both credential fields or omit both to use the default AWS credential provider chain.
+- **store_region**: The region of the object storage. Required, including for S3-compatible stores that ignore it.
- **store_class**: The storage class to use. **Currently, only S3-compatible storage is supported.**
- **store_path_style_access**: Use path-style S3 URLs (`http://host/bucket/key`). Defaults to `true`, which MinIO-style endpoints require; set to `false` for stores that only accept virtual-hosted-style URLs.
+All options above are required except the credential pair and `store_path_style_access`.
+`tables` and `dynamic_route_field` must be present even when the selected mode does not use them.
+
+## Static Routing
+
+With `dynamic_routing = false`, every batch is copied to every successfully loaded table in
+`tables`. Invalid names and tables that cannot be loaded are skipped at startup. Startup fails
+if none can be loaded. The writer uses the schema and default partition spec captured at
+startup, so restart the connector after changing them.
+
## Dynamic Routing
If you don't know the names of the Iceberg tables you want to route data to in advance, you can use the dynamic routing feature.
-Insert a field in your Iggy messages with the name of the Iceberg table the message should be routed to. The Iggy connector will parse this field at runtime and route the message to the correct table.
+Insert a top-level field in your JSON messages with the name of the Iceberg table the message should be routed to. The Iggy connector will parse this field at runtime and route the message to the correct table.
The Iggy Iceberg Connector will skip messages in the following cases:
-- The table declared in the message field does not exist.
-- The message does not contain the field specified in the `dynamic_route_field` configuration option.
+- The table declared in the message field cannot be loaded, including missing tables and catalog lookup failures.
+- The table name has no namespace or contains an empty name component.
+- The message does not contain the field specified in the `dynamic_route_field` configuration option, or is not a JSON object.
+
+A lookup failure skips the affected message without failing the batch. Each batch loads its
+destination tables again. The route field remains in the row, but the writer ignores it if the
+target schema has no matching column.
### Dynamic routing configuration example
```toml
[plugin_config]
-tables = [""]
+tables = []
catalog_type = "rest"
warehouse = "warehouse"
uri = "http://localhost:8181"
@@ -70,10 +89,10 @@ store_region = "us-east-1"
store_class = "s3"
store_path_style_access = true
-[sinks.iceberg.transforms.add_fields]
+[transforms.add_fields]
enabled = true
-[[sinks.iceberg.transforms.add_fields.fields]]
+[[transforms.add_fields.fields]]
key = "db_table"
value.static = "nyc.users"
```
@@ -88,15 +107,25 @@ Example:
Data files follow the table's default partition spec. Each batch is split by the spec's transforms
(`identity`, `bucket`, `truncate`, `year`, `month`, `day`, `hour`) and every partition value present
-in the batch gets at least one Parquet file under its partition path. Unpartitioned tables are
-written into a single file per batch.
+in a successful batch gets at least one Parquet file under its partition path. Writers can roll
+over into multiple files, including for unpartitioned tables.
+
+Each destination table has its own append transaction. Fan-out is not atomic across tables:
+a write or commit error stops work on the remaining tables, while earlier commits remain.
+The Iceberg library refreshes metadata before committing and retries eligible commit errors
+according to the table's retry properties. The plugin does not replay failed batches. The
+runtime records a plugin error and continues polling with consumer auto-commit, so messages
+can be lost after failed or skipped writes. Replaying committed rows can create duplicates.
## Source Compatibility
-The Iceberg sink expects **flat JSON** where each top-level key maps directly to a column in the
-target Iceberg table schema. Sources that wrap row data in an envelope (with metadata fields
-alongside a nested data object) are not directly compatible — the Arrow JSON reader will map
-envelope keys to table columns, producing nulls or schema errors.
+Use `schema = "json"` on the stream. Each JSON object represents a table row; nested objects
+and arrays are supported when they match the table schema. Unknown fields are ignored. Missing
+nullable fields become null; missing required fields or incompatible values fail the table write.
+
+Sources that wrap row data in an envelope need a transform when the table schema describes the
+inner row. Otherwise, the Arrow JSON reader maps envelope keys to table columns, producing nulls
+or schema errors.
If your source emits envelope-wrapped JSON, use the `unwrap_envelope` transform to extract the
inner data field before it reaches the sink:
diff --git a/core/connectors/sinks/influxdb_sink/README.md b/core/connectors/sinks/influxdb_sink/README.md
index 64813f8d8..1ab1b5b91 100644
--- a/core/connectors/sinks/influxdb_sink/README.md
+++ b/core/connectors/sinks/influxdb_sink/README.md
@@ -9,16 +9,21 @@ A sink connector that consumes messages from Iggy streams and writes them to Inf
| Data organisation | `org` + `bucket` | `db` |
| Write endpoint | `POST /api/v2/write` | `POST /api/v3/write_lp` |
| Auth header | `Authorization: Token {t}` | `Authorization: Bearer {t}` |
-| Precision values | `ns`, `us`, `ms`, `s` | same short forms accepted |
-| Config `version` key | `"v2"` (or omit — default) | `"v3"` |
+| Write API precision values | `ns`, `us`, `ms`, `s` | `nanosecond`, `microsecond`, `millisecond`, `second` (mapped from connector short forms) |
+| Config `version` key | `"v2"` (or omit, default) | `"v3"` |
-The write body (InfluxDB line protocol), retry/circuit-breaker behaviour, batch accumulation, and payload format handling are identical between versions.
+The write body (InfluxDB line protocol), retry/circuit-breaker behaviour, batch chunking, and payload format handling are identical between versions.
## Configuration
Select the version with `version = "v2"` or `version = "v3"`. Omitting `version` defaults to `"v2"` for backward compatibility with existing deployments.
+Unknown options are rejected, so remove `org` and `bucket` when switching to V3.
+The examples in the next sections are the contents of `[plugin_config]`; the full connector
+file appears below. The target organization/bucket or database must be configured and the
+token must have write access. Startup probes `GET /health` with the configured token and version-specific authentication
+header. A successful health check does not validate write permissions.
-### V2 — InfluxDB OSS 2.x / Cloud
+### V2 - InfluxDB OSS 2.x / Cloud
```toml
version = "v2"
@@ -33,7 +38,7 @@ precision = "us" # ns | us | ms | s (default: "us")
batch_size = 500 # messages per write request (default: 500)
payload_format = "json" # json | text | base64 (default: "json")
-include_metadata = true # inject offset/stream/topic fields (default: true)
+include_metadata = true # include stream/topic/partition as tags or fields (default: true)
include_checksum = false
include_origin_timestamp = false
include_stream_tag = false # add stream name as a line-protocol tag
@@ -42,7 +47,7 @@ include_partition_tag = false # add partition id as a line-protocol tag
verbose_logging = false
```
-### V3 — InfluxDB 3.x Core / Enterprise
+### V3 - InfluxDB 3.x Core / Enterprise
```toml
version = "v3"
@@ -50,7 +55,7 @@ url = "http://localhost:8181"
db = "my-db"
token = "my-token"
-# Optional — same fields as V2 except org/bucket are replaced by db
+# Optional - same fields as V2 except org/bucket are replaced by db
measurement = "iggy_events"
precision = "us"
batch_size = 500
@@ -69,37 +74,62 @@ verbose_logging = false
```toml
timeout = "30s" # per-request timeout
-max_retries = 3 # total write attempts, including the first (429/5xx)
+max_retries = 3 # total write attempts, including the first (network errors/429/5xx)
retry_delay = "1s" # initial backoff between retries
retry_max_delay = "5s" # backoff cap
max_open_retries = 10 # total open() health-check attempts, including the first
open_retry_max_delay = "60s" # backoff cap for open() retries
circuit_breaker_threshold = 5 # consecutive failures before circuit trips
-circuit_breaker_cool_down = "30s" # how long circuit stays open before half-open probe
+circuit_breaker_cool_down = "30s" # how long circuit stays open before writes resume
```
## Payload Formats
-- **`json`** (default): Each message payload is parsed as JSON and its fields become line-protocol field set entries.
-- **`text`**: Payload treated as a plain string; written as a single `value` field.
-- **`base64`**: Payload decoded from base64; written as raw bytes (stored as a string field).
+- **`json`** (default): Each message payload is validated as JSON, compact-serialized, and stored in one `payload_json` string field.
+- **`text`**: Payload must be valid UTF-8 and is stored in one `payload_text` string field.
+- **`base64`**: Payload bytes are base64-encoded into one `payload_base64` string field.
+
+Formats are case-insensitive; `utf8` aliases `text`, `raw` aliases `base64`, and an unknown
+format warns and falls back to JSON. Formatting happens after stream decoding and transforms.
+Use `schema = "raw"` with base64 and no payload-changing transform to preserve arbitrary bytes,
+or `schema = "text"` with text for UTF-8 input. Text CR/LF bytes become literal `\r`/`\n`
+sequences in the stored string. Measurement names and tag values reject tabs; text fields allow them.
+
+## Metadata and timestamps
+
+All `include_*` options default to `true`. `message_id` is always a string field and `offset`
+is always a tag. With `include_metadata = true`, stream/topic/partition are tags unless their
+individual tag flags are disabled, in which case they become `iggy_stream`, `iggy_topic` and
+`iggy_partition` fields. `include_metadata = false` omits those three values. Checksum and
+origin timestamp have independent flags.
+
+Point timestamps use the message timestamp, converted from microseconds to the configured
+precision. Zero uses the current wall-clock time. Millisecond/second conversion truncates
+finer digits. InfluxDB point identity uses measurement/table, tags and timestamp; removing
+stream/topic/partition tags can collapse distinct messages with the same offset and timestamp.
+Moving those values to fields does not preserve identity. See the
+[V2](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#duplicate-points) and
+[V3](https://docs.influxdata.com/influxdb3/core/reference/line-protocol/#duplicate-points) duplicate-point rules.
## Full Configuration Example
```toml
-[[sinks]]
-key = "influxdb-sink"
+type = "sink"
+key = "influxdb"
enabled = true
-path = "target/release/libiggy_connector_influxdb_sink"
-
-[[sinks.streams]]
-stream = "metrics"
-topic = "cpu"
-schema = "json"
+version = 0
+name = "InfluxDB sink"
+path = "target/release/libiggy_connector_influxdb_sink"
+
+[[streams]]
+stream = "metrics"
+topics = ["cpu"]
+schema = "json"
batch_length = 100
-linger_time = "10ms"
+poll_interval = "10ms"
+consumer_group = "influxdb_sink"
-[sinks.plugin_config]
+[plugin_config]
version = "v2"
url = "http://localhost:8086"
org = "acme"
@@ -118,7 +148,18 @@ circuit_breaker_cool_down = "15s"
The sink uses a layered design:
-- **Batch accumulator**: messages are serialised to line protocol and buffered until `batch_size` is reached, then flushed in a single HTTP POST.
+- **Batch chunking**: each polled batch is split into requests of at most `batch_size` messages (`0` behaves as `1`). The final partial chunk is written immediately; messages are not accumulated across polls.
- **Retry middleware**: `iggy_connector_sdk::retry::HttpRetryMiddleware` retries 429, 5xx and network errors with exponential backoff and jitter.
-- **Circuit breaker**: after `circuit_breaker_threshold` consecutive failures the connector stops issuing writes and waits for the cool-down window before probing again.
+- **Circuit breaker**: records at most one failure per consume call, based on its first error. Permanent HTTP errors do not increment the counter; a fully successful call resets it. While open, batches fail without writes. After the cool-down window, writes resume and the counter resets.
- **Precision mapping**: V3's `/api/v3/write_lp` endpoint requires full English words (`nanosecond`, `microsecond`, `millisecond`, `second`); the connector maps the short forms automatically.
+
+A serialization error rejects its whole chunk. Later chunks are still attempted after failures,
+and the first error is returned after the loop. `max_retries = 3` allows two retries; `0` and
+`1` both allow one attempt. Backoff starts at `retry_delay`, doubles with ±20% jitter and is
+capped by `retry_max_delay`. Integer-seconds `Retry-After` on a 429 overrides that cap;
+HTTP-date values are ignored. Startup uses its own attempt budget/cap and retries any failed
+health check. Invalid duration strings warn and fall back to `1s`.
+
+The runtime records a plugin error and continues polling with consumer auto-commit. Failed
+chunks and batches skipped by the circuit breaker are not queued for replay. Other chunks
+may already be stored, so these mechanisms do not provide an end-to-end at-least-once guarantee.
diff --git a/core/connectors/sinks/influxdb_sink/src/lib.rs b/core/connectors/sinks/influxdb_sink/src/lib.rs
index 2018bd90f..cef1fd609 100644
--- a/core/connectors/sinks/influxdb_sink/src/lib.rs
+++ b/core/connectors/sinks/influxdb_sink/src/lib.rs
@@ -29,6 +29,7 @@ use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
};
use reqwest::Url;
+use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
use reqwest_middleware::ClientWithMiddleware;
use secrecy::{ExposeSecret, SecretBox, SecretString};
use serde::{Deserialize, Serialize};
@@ -454,7 +455,15 @@ impl InfluxDbSink {
fn build_raw_client(&self) -> Result<reqwest::Client, Error> {
let timeout = parse_duration(self.config.timeout(), DEFAULT_TIMEOUT);
+ let mut authorization =
+ HeaderValue::from_str(&self.config.auth_header()).map_err(|error| {
+ Error::InvalidConfigValue(format!("Invalid InfluxDB authorization header: {error}"))
+ })?;
+ authorization.set_sensitive(true);
+ let mut headers = HeaderMap::new();
+ headers.insert(AUTHORIZATION, authorization);
reqwest::Client::builder()
+ .default_headers(headers)
.timeout(timeout)
.build()
.map_err(|e| Error::InitError(format!("Failed to create HTTP client: {e}")))
@@ -503,10 +512,8 @@ impl InfluxDbSink {
if self.include_metadata && self.include_partition_tag {
write!(buf, ",partition={}", messages_metadata.partition_id).expect("infallible");
}
- // `offset` is always written as a tag regardless of `include_metadata`.
- // It forms the deduplication key for idempotent writes: without it, two
- // messages at the same timestamp in the same measurement+tag-set would
- // silently overwrite each other in InfluxDB's last-write-wins model.
+ // Offset keeps messages in one partition distinct at the same timestamp.
+ // Removing stream/topic/partition tags can still collapse their identities.
write!(buf, ",offset={}", message.offset).expect("infallible");
buf.push(' ');
@@ -1962,6 +1969,36 @@ mod http_tests {
assert!(sink.open().await.is_ok());
}
+ #[tokio::test]
+ async fn open_authenticates_health_check() {
+ for (is_v3, expected_auth) in [(true, "Bearer tok"), (false, "Token tok")] {
+ let app = Router::new().route(
+ "/health",
+ get(move |headers: HeaderMap| async move {
+ if headers
+ .get("authorization")
+ .and_then(|value| value.to_str().ok())
+ == Some(expected_auth)
+ {
+ StatusCode::OK
+ } else {
+ StatusCode::UNAUTHORIZED
+ }
+ }),
+ );
+ let base = start_server(app).await;
+ let config = if is_v3 {
+ v3_config(&base)
+ } else {
+ v2_config(&base)
+ };
+ let mut sink = InfluxDbSink::new(1, config);
+ sink.open()
+ .await
+ .expect("health check must use the configured version's authorization header");
+ }
+ }
+
#[tokio::test]
async fn open_fails_when_health_returns_503() {
let app = Router::new().route("/health", get(|| async { StatusCode::SERVICE_UNAVAILABLE }));
@@ -2233,11 +2270,8 @@ mod http_tests {
}
#[tokio::test]
- async fn consume_records_success_per_successful_batch() {
- // With 2 batches where the first fails and the second succeeds, the circuit
- // breaker must record 1 failure AND 1 success — not 1 failure and 0 successes.
- // If only failures are recorded, the breaker will trip after enough intermittent
- // errors even when most batches succeed.
+ async fn consume_stays_closed_below_failure_threshold() {
+ // One failed consume call remains below the configured threshold of two.
let call_count = Arc::new(AtomicU32::new(0));
let cc2 = call_count.clone();
let app = Router::new()
@@ -2257,9 +2291,6 @@ mod http_tests {
}),
);
let base = start_server(app).await;
- // Use a threshold of 2 so the breaker trips after 2 failures.
- // If the success of batch 2 is not recorded, successive calls that have
- // 1 failure each would trip the breaker after 2 invocations.
let config = InfluxDbSinkConfig::V2(V2SinkConfig {
circuit_breaker_threshold: Some(2),
circuit_breaker_cool_down: Some("60s".to_string()),
@@ -2268,10 +2299,9 @@ mod http_tests {
let sink = open_sink(config).await;
let msgs: Vec<_> = (0..4).map(|_| msg()).collect();
let _ = sink.consume(&topic(), meta(), msgs).await; // first fails, second succeeds
- // Circuit breaker should NOT be open: 1 failure + 1 success → not tripped.
assert!(
!sink.circuit_breaker.is_open().await,
- "circuit breaker must not trip when at least one batch succeeded"
+ "one failed consume call must remain below the threshold of two"
);
}
diff --git a/core/connectors/sinks/influxdb_sink/src/protocol.rs b/core/connectors/sinks/influxdb_sink/src/protocol.rs
index 80cd20f3c..ab0d8d197 100644
--- a/core/connectors/sinks/influxdb_sink/src/protocol.rs
+++ b/core/connectors/sinks/influxdb_sink/src/protocol.rs
@@ -29,8 +29,7 @@ use iggy_connector_sdk::Error;
/// Newline and carriage-return are the InfluxDB line-protocol record
/// delimiters that can corrupt parsing; a literal newline inside
/// a measurement name would split the line and corrupt the batch.
-/// Tab characters are rejected with an error as they are not valid
-/// in the InfluxDB line-protocol spec and would corrupt tag-set parsing.
+/// This connector rejects tab characters in unquoted names.
#[inline]
pub(crate) fn write_measurement(buf: &mut String, value: &str) -> Result<(), Error> {
// Span-copy: all escape targets are ASCII (single byte), so byte positions
@@ -42,9 +41,7 @@ pub(crate) fn write_measurement(buf: &mut String, value: &str) -> Result<(), Err
let esc = match b {
b'\t' => {
return Err(Error::InvalidConfigValue(
- "measurement name must not contain tab characters — tabs are not valid \
- in the InfluxDB line-protocol spec and would corrupt tag-set parsing"
- .into(),
+ "measurement name must not contain tab characters".into(),
));
}
b'\\' => "\\\\",
@@ -68,8 +65,7 @@ pub(crate) fn write_measurement(buf: &mut String, value: &str) -> Result<(), Err
///
/// Newline and carriage-return are InfluxDB line-protocol record delimiters that
/// can corrupt tag-set parsing.
-/// Tab characters are rejected with an error as they are not valid
-/// in the InfluxDB line-protocol spec and would corrupt tag-set parsing.
+/// This connector rejects tab characters in unquoted names.
#[inline]
pub(crate) fn write_tag_value(buf: &mut String, value: &str) -> Result<(), Error> {
// Span-copy: all escape targets are ASCII (single byte), so byte positions
@@ -81,9 +77,7 @@ pub(crate) fn write_tag_value(buf: &mut String, value: &str) -> Result<(), Error
let esc = match b {
b'\t' => {
return Err(Error::CannotStoreData(
- "tag value must not contain tab characters — tabs are not valid \
- in the InfluxDB line-protocol spec and would corrupt tag-set parsing"
- .into(),
+ "tag value must not contain tab characters".into(),
));
}
b'\\' => "\\\\",
@@ -106,18 +100,10 @@ pub(crate) fn write_tag_value(buf: &mut String, value: &str) -> Result<(), Error
///
/// Escapes: `\` → `\\`, `"` → `\"`, `\n` → `\\n`, `\r` → `\\r`
///
-/// NOTE: Escaping `\n` and `\r` inside quoted string field values is a
-/// deliberate divergence from the InfluxDB line-protocol spec (which only
-/// mandates `\\` and `\"` inside quoted strings). The divergence prevents
-/// line-splitting in downstream consumers that use raw newlines as record
-/// delimiters — e.g. log shippers, stream processors, and InfluxDB's own
-/// bulk-write HTTP parser if the body is re-streamed line by line.
-///
-/// Tab (`\t`) is intentionally NOT escaped here. String field values are
-/// double-quoted in line protocol, and the spec permits literal tabs inside
-/// quoted strings. Measurement names and tag values (see [`write_measurement`]
-/// and [`write_tag_value`]) are unquoted, so tabs must be escaped there to
-/// avoid misparsing the tag set.
+/// Line protocol does not support literal newlines in field values. This
+/// connector stores CR/LF as visible `\r`/`\n` text, which does not preserve
+/// those original bytes. Use the base64 payload format when they must round-trip.
+/// Literal tabs are allowed in quoted string fields.
#[inline]
pub(crate) fn write_field_string(buf: &mut String, value: &str) {
// Span-copy: all escape targets are ASCII (single byte), so byte positions
diff --git a/core/connectors/sinks/meilisearch_sink/README.md b/core/connectors/sinks/meilisearch_sink/README.md
index f164c83dd..8545198c6 100644
--- a/core/connectors/sinks/meilisearch_sink/README.md
+++ b/core/connectors/sinks/meilisearch_sink/README.md
@@ -30,14 +30,16 @@ max_open_retries = 5
- `api_key`: Optional Meilisearch API key sent as `Authorization: Bearer`.
Use HTTPS for non-local Meilisearch hosts; HTTP sends the key without
transport encryption.
-- `primary_key`: Index primary key field. Defaults to `iggy_id`.
+- `primary_key`: Index primary key field. Defaults to `iggy_id`. Match the
+ existing index key: a mismatch only warns at startup and can fail writes.
- `document_action`: `replace` uses SDK add-or-replace semantics; `update`
uses SDK add-or-update semantics. Defaults to `replace`.
- `create_index_if_not_exists`: Create the index during `open()` when missing. Defaults to `true`.
- `include_metadata`: Add Iggy metadata fields to each document. Defaults to `true`.
- `batch_size`: Maximum documents per Meilisearch document request. Defaults to `1000`.
-- `timeout`: Request timeout as a humantime string, for example `30s`. Defaults to `30s`.
-- `wait_for_tasks`: Poll Meilisearch tasks until terminal state before
+- `timeout`: Total deadline for one SDK operation including retries and backoff.
+ Health checks apply it separately to each attempt. Defaults to `30s`.
+- `wait_for_tasks`: Wait for indexing success or failure, bounded by `task_timeout`, before
returning from `consume()`. Defaults to `true`. Setting this to `false` makes
document indexing fire-and-forget, so asynchronous Meilisearch task failures
are not observed by the connector.
@@ -49,7 +51,7 @@ max_open_retries = 5
- `max_open_retries`: Maximum transient retries after the initial request while
opening the index. Defaults to `5`. This also applies to `get_task` polls
while waiting for index creation during `open()`. Each retried open operation
- uses the configured request `timeout` plus backoff; there is no single total
+ uses one `timeout` deadline including backoff; there is no single total
open deadline.
## Behavior
@@ -58,16 +60,17 @@ JSON object payloads are indexed as documents. JSON arrays or scalar values are
wrapped in a `value` field because Meilisearch documents must be objects. Raw
payloads are parsed as JSON when possible; otherwise, they are indexed as base64
data. Text payloads are indexed in a `text` field. Unsupported payload schemas
-are skipped with a warning and counted as sink errors, matching the connector
-runtime's per-record drop behavior for malformed records. Because the sink
-returns success after dropping an unsupported-schema record, the runtime can
-commit the consumer offset for that record. There is no built-in dead-letter
-queue for these drops.
+are skipped with a warning and counted in the plugin's private error counter.
+The callback returns success after these drops, so runtime statistics can count
+those records as processed. Offsets are auto-committed while polling, before
+indexing completes. There is no built-in dead-letter queue for these drops.
When the configured primary key is absent, the connector injects a stable value
derived from the exact Iggy stream, topic, partition, offset, and message ID.
This avoids Meilisearch primary-key inference failures. If the payload already
-contains the configured primary key, that value is preserved. Operators must
+contains the configured primary key, that value is preserved unless a reserved
+metadata field below overwrites it. A present but invalid or null key is not
+replaced automatically. Operators must
ensure user-provided primary keys are unique, otherwise Meilisearch
add-or-replace semantics can collapse distinct messages into one document.
@@ -77,29 +80,42 @@ payload fields so audit metadata reflects the actual stream, topic, partition,
offset, checksum, and timestamps. `iggy_checksum` is stored as a string to avoid
JSON number precision loss in Meilisearch clients. Offset and timestamp metadata
remain JSON numbers. If `primary_key` is set to a field other than `iggy_id`,
-the connector also writes `iggy_id` as stable Iggy metadata.
+the connector also writes `iggy_id` as stable Iggy metadata. A supplied `iggy_id`
+is preserved when it is the configured primary key; avoid other reserved
+metadata fields as a primary key because metadata overwrites them. Message
+and origin timestamps are microseconds; `iggy_ingested_at` uses wall-clock
+milliseconds. Disabling metadata leaves payload fields intact and still injects
+a missing primary key.
## Delivery Semantics
The connector runtime invokes `consume()` through an FFI callback whose status
code is not currently used to gate offset commits. A batch error returned by the
sink is logged by the sink, but the runtime does not redeliver that batch. The
-effective runtime-level delivery guarantee is at-most-once on sink errors. The
-sink's retry settings only provide best-effort retries inside a single
-`consume()` call.
+runtime records a plugin error and continues polling. There is no end-to-end
+at-least-once guarantee; retries or manual replay after uncertain task outcomes
+can repeat writes. The sink retries transient submission/status requests within
+a single `consume()` call. Failed tasks are not resubmitted, and task timeouts
+do not cancel remote tasks.
`wait_for_tasks=false` only skips waiting for document indexing tasks during
-`consume()`. In that mode, successful submission lets the runtime commit the
-consumer offset before Meilisearch has confirmed indexing, so later task
+`consume()`. In that mode, submission succeeds before Meilisearch confirms
+indexing, while offsets have already been auto-committed during polling, so later task
failures are not retried, logged, or counted by this connector. If
`create_index_if_not_exists=true` and the connector creates the index during
`open()`, it still waits for that index-creation task so the first batch cannot
race the index creation. This mode is fire-and-forget and does not provide
-durability.
+durability. Closing does not await outstanding document tasks.
-The close-time counters are attempt counters. `documents_enqueued` counts
-documents accepted by completed Meilisearch SDK calls in this process, and
-`documents_confirmed` counts those same documents only when task waiting is
-enabled and the corresponding task reached success. `errors` includes invalid
+Each polled batch is chunked immediately, including its final partial chunk;
+there is no accumulation across polls. `batch_size = 0` behaves as `1`. A failed
+chunk stops the loop, so earlier chunks may be stored while later chunks are
+never attempted.
+
+The close-time counters cover this process. `documents_enqueued` counts
+successfully handled chunks: submission must succeed and, when task waiting
+is enabled, the task must also succeed. `documents_confirmed` counts those same
+documents only with task waiting enabled. Submitted tasks that fail or time out
+are excluded from both success counters, even if a timed-out task later succeeds. `errors` includes invalid
records plus documents in failed chunks and trailing chunks that were not
attempted after an earlier chunk failed.
diff --git a/core/connectors/sinks/mongodb_sink/README.md b/core/connectors/sinks/mongodb_sink/README.md
index 5e0dd4cfa..05f3d4858 100644
--- a/core/connectors/sinks/mongodb_sink/README.md
+++ b/core/connectors/sinks/mongodb_sink/README.md
@@ -10,14 +10,14 @@ Send a JSON message through Iggy and see it land in MongoDB.
```bash
# Start MongoDB
-docker run -d --name mongo-test -p 27017:27017 mongo:7
+docker run -d --name mongo-test -p 127.0.0.1:27017:27017 mongo:7
# Start iggy-server (terminal 2)
-IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy ./target/debug/iggy-server
+IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy ./target/debug/iggy-server --replica-id 0
# Create stream and topic
./target/debug/iggy -u iggy -p iggy stream create demo_stream
-./target/debug/iggy -u iggy -p iggy topic create demo_stream demo_topic 1
+./target/debug/iggy -u iggy -p iggy topic create demo_stream demo_topic 1 none 1d
# Setup connector config
mkdir -p /tmp/mdb-sink-test/connectors
@@ -58,14 +58,14 @@ TOML
IGGY_CONNECTORS_CONFIG_PATH=/tmp/mdb-sink-test/config.toml ./target/debug/iggy-connectors
# Send a message
-./target/debug/iggy -u iggy -p iggy message send demo_stream demo_topic '{"hello":"mongodb"}'
+./target/debug/iggy -u iggy -p iggy message send --partition-id 0 demo_stream demo_topic '{"hello":"mongodb"}'
# Verify in MongoDB
docker exec mongo-test mongosh --quiet --eval \
'db.getSiblingDB("test_db").messages.find().pretty()'
```
-Expected:
+Output includes:
```json
{ "payload": { "hello": "mongodb" }, "iggy_offset": 0, "iggy_stream": "demo_stream" }
@@ -98,15 +98,15 @@ payload_format = "json"
| `connection_uri` | **required** | MongoDB URI |
| `database` | **required** | Target database |
| `collection` | **required** | Target collection |
-| `batch_size` | `100` | Documents per `insertMany` call |
+| `batch_size` | `100` | Maximum documents per `insertMany` call; `0` behaves as `1` |
| `payload_format` | `binary` | `binary`, `json`, or `string` |
| `include_metadata` | `true` | Add iggy offset, timestamp, stream, topic, partition |
| `include_checksum` | `true` | Add message checksum |
| `include_origin_timestamp` | `true` | Add origin timestamp |
-| `auto_create_collection` | `false` | Create collection if missing |
+| `auto_create_collection` | `false` | Explicitly create a missing collection at startup; `false` still permits implicit creation by an insert |
| `max_pool_size` | driver default | Connection pool size |
| `verbose_logging` | `false` | Log at info instead of debug |
-| `max_retries` | `3` | Retry attempts for transient errors |
+| `max_retries` | `3` | Total attempts for transient insert errors, including the first; `0` behaves as `1` |
| `retry_delay` | `1s` | Base delay (`retry_delay * attempt`) |
## Testing
@@ -114,15 +114,16 @@ payload_format = "json"
Requires Docker. Testcontainers starts MongoDB 7 + iggy-server automatically.
```bash
-cargo test --test mod -- mongodb_sink
+cargo nextest run -p integration -E 'test(connectors::mongodb::)'
```
-This runs 4 E2E tests against a real MongoDB instance:
+The integration suite covers payloads, batching, duplicate handling, validation failures,
+write concern and retryable writes against real MongoDB instances. Cases include:
-- `json_messages_sink_to_mongodb` — JSON payloads stored as embedded BSON documents
-- `binary_messages_sink_as_bson_binary` — binary payloads stored as BSON Binary
-- `large_batch_processed_correctly` — batch insertion with configurable batch size
-- `auto_create_collection_on_open` — collection created automatically when missing
+- `json_messages_sink_to_mongodb` - JSON payloads stored as embedded BSON documents
+- `binary_messages_sink_as_bson_binary` - binary payloads stored as BSON Binary
+- `large_batch_processed_correctly` - batch insertion with configurable batch size
+- `auto_create_collection_on_open` - collection created automatically when missing
Unit tests (no Docker):
@@ -132,16 +133,41 @@ cargo test -p iggy_connector_mongodb_sink
## Delivery Semantics
-This connector provides **at-least-once** delivery semantics.
+The runtime auto-commits while polling, records plugin errors and continues
+without replaying failed batches. This does not provide an end-to-end
+at-least-once guarantee.
### Behavior
-- Messages may be delivered more than once on retry or restart
-- Uses a deterministic composite MongoDB `_id`: `stream:topic:partition:message_id`
-- Duplicate key collisions are treated as idempotent replay of already-written messages
-- The sink remains insert-only; it does not upsert existing documents
-
-### Known Limitations
-
-- On network timeout during insert, MongoDB may partially commit a batch before returning an error
-- The sink does not upsert on duplicate; replay safety relies on deterministic `_id` values and duplicate-key tolerance
+- Each document has a deterministic `_id`: `stream:topic:partition:message_id`.
+ Offset is not included; reuse of an ID can discard a different message.
+- Payloads remain under `payload`, as BSON Binary, a JSON-derived BSON value,
+ or a UTF-8 string. JSON can be an object, array or scalar; invalid conversion
+ rejects its entire chunk. Formatting follows stream decoding and transforms.
+- Message headers are not stored. Checksum and origin timestamp flags are
+ independent of `include_metadata`. BSON datetimes truncate microseconds to
+ milliseconds; large offsets use `iggy_offset_str`, and large checksums use
+ decimal strings when they do not fit a signed 64-bit integer.
+- Current-poll chunks are inserted immediately with unordered `insert_many`.
+ Later chunks are still attempted after errors; the last chunk error is
+ returned. There is no cross-poll buffer or batch-wide transaction.
+- All-duplicate write errors (`11000`) without a write-concern error are treated
+ as success. This applies to every unique index, not only `_id`, without
+ comparing payload contents. A conflict on another unique index can discard
+ a distinct message while runtime statistics report it processed.
+- The sink is insert-only and does not update existing documents.
+
+### Retry and failure limits
+
+`max_retries` counts total plugin attempts. Linear backoff waits `retry_delay`
+multiplied by the retry number; an invalid duration silently falls back to `1s`.
+Driver-level retryable writes can retry independently, depending on the URI and
+deployment. Startup parses the URI, pings the target database and optionally
+creates the collection, without a connector-level retry loop. An explicit
+`max_pool_size` overrides the URI pool setting.
+
+MongoDB can store part of an unordered insert before returning an error.
+Timeouts and write-concern failures can leave the outcome uncertain. The
+plugin's inserted-message counter reports known or estimated insert counts;
+its error counter counts failed chunks. Runtime counters cover whole callbacks,
+so they are not exact MongoDB document counts.
diff --git a/core/connectors/sinks/mongodb_sink/src/lib.rs b/core/connectors/sinks/mongodb_sink/src/lib.rs
index 1c4f37835..4cf24b4ce 100644
--- a/core/connectors/sinks/mongodb_sink/src/lib.rs
+++ b/core/connectors/sinks/mongodb_sink/src/lib.rs
@@ -835,15 +835,6 @@ mod tests {
assert!(!is_transient_error(&e));
}
- // ---- process_messages error propagation tests ----
- // These tests verify that the sink does NOT silently lose data when inserts fail.
-
- /// Test contract: When MongoDB insert fails, process_messages MUST return Err.
- /// This prevents silent data loss where upstream commits while writes failed.
- ///
- /// Given: A sink with no client (will fail on get_client)
- /// When: process_messages is called with messages
- /// Then: Returns Err (not Ok) and does NOT count failed messages as processed
#[tokio::test]
async fn given_no_client_should_return_error_not_silent_ok() {
let config = given_default_config();
diff --git a/core/connectors/sinks/postgres_sink/README.md b/core/connectors/sinks/postgres_sink/README.md
index 5a4884acb..1d20d764a 100644
--- a/core/connectors/sinks/postgres_sink/README.md
+++ b/core/connectors/sinks/postgres_sink/README.md
@@ -13,7 +13,26 @@ The PostgreSQL sink connector consumes messages from Iggy topics and stores them
## Configuration
+Build from the matching 0.9.0/edge checkout root:
+
+```bash
+cargo build --release -p iggy_connector_postgres_sink
+```
+
+Configure the broker connection and connector directory as described in the
+[runtime README](../../runtime/README.md). Save the following connector file
+there. Paths are relative to the process working directory, here the checkout
+root. The URI assumes a local `iggy` database and `iggy`/`iggy` credentials;
+replace it for your deployment.
+
```toml
+type = "sink"
+key = "postgres-sink"
+enabled = true
+version = 1
+name = "Postgres Sink"
+path = "target/release/libiggy_connector_postgres_sink"
+
[[streams]]
stream = "user_events"
topics = ["users", "orders"]
@@ -23,7 +42,7 @@ poll_interval = "5ms"
consumer_group = "postgres_sink"
[plugin_config]
-connection_string = "postgresql://user:pass@localhost:5432/database"
+connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
target_table = "iggy_messages"
batch_size = 100
max_connections = 10
@@ -39,17 +58,17 @@ payload_format = "bytea"
| Option | Type | Default | Description |
| ------ | ---- | ------- | ----------- |
| `connection_string` | string | required | PostgreSQL connection string |
-| `target_table` | string | required | Target table name |
-| `batch_size` | u32 | `100` | Messages per insert batch |
+| `target_table` | string | required | One quoted table identifier; `schema.table` is a literal name containing a dot |
+| `batch_size` | u32 | `100` | Maximum messages per insert statement; `0` behaves as `1` |
| `max_connections` | u32 | `10` | Max database connections |
-| `auto_create_table` | bool | `false` | Create table if not exists |
+| `auto_create_table` | bool | `false` | Create a missing table; existing tables are not migrated |
| `include_metadata` | bool | `true` | Include Iggy metadata columns |
| `include_checksum` | bool | `true` | Include message checksum |
| `include_origin_timestamp` | bool | `true` | Include original timestamp |
-| `payload_format` | string | `bytea` | Payload column type: `bytea`, `json`, or `text` |
+| `payload_format` | string | `bytea` | Payload column type: `bytea`, `json` (alias `jsonb`), or `text` |
| `verbose_logging` | bool | `false` | Log at info level instead of debug |
-| `max_retries` | u32 | `3` | Max retry attempts for transient errors |
-| `retry_delay` | string | `1s` | Base delay between retries (e.g., `500ms`, `2s`) |
+| `max_retries` | u32 | `3` | Total attempts per transiently failing insert; `0` and `1` both allow one attempt |
+| `retry_delay` | string | `1s` | Base for linear retry delays; invalid values silently fall back to `1s` |
## Payload Format
@@ -57,13 +76,20 @@ The `payload_format` option determines how the payload is stored in PostgreSQL:
| Format | Column Type | Description |
| ------ | ----------- | ----------- |
-| `bytea` | `BYTEA` | Raw bytes (default). Preserves exact binary content. |
-| `json` / `jsonb` | `JSONB` | Native JSON. Enables JSON queries and indexing. Payload must be valid JSON. |
-| `text` | `TEXT` | UTF-8 text. Payload must be valid UTF-8. |
+| `bytea` | `BYTEA` | Bytes after stream decoding and transforms (default). |
+| `json` / `jsonb` | `JSONB` | Native JSON. Enables JSON queries and indexing. Payload must be valid JSON accepted by PostgreSQL JSONB. |
+| `text` | `TEXT` | UTF-8 text. Payload must be valid UTF-8 accepted by PostgreSQL TEXT. |
+
+Format names are case-insensitive; unknown names silently use BYTEA. JSONB
+accepts objects, arrays and scalars but rejects `\u0000`; TEXT rejects zero bytes.
+Headers are not stored. The following format snippets change the corresponding
+setting in the complete connector file above.
### BYTEA (Default)
-Stores the exact bytes from the Iggy message. Use for binary data, protobuf, or when you want to preserve the original format.
+Stores the bytes after decoding and transforms. Use `schema = "raw"` without
+payload-changing transforms to preserve original binary or protobuf bytes.
+JSON decoding can change the byte representation before the sink receives it.
```toml
[plugin_config]
@@ -106,7 +132,8 @@ SELECT id, payload FROM iggy_messages WHERE payload LIKE '%error%';
## Table Schema
-When `auto_create_table` is enabled, the table is created with the appropriate payload column type:
+With `auto_create_table = true` and all three metadata flags enabled, a missing
+table gets the following structure. The payload type follows `payload_format`:
```sql
CREATE TABLE iggy_messages (
@@ -123,10 +150,27 @@ CREATE TABLE iggy_messages (
);
```
+Disabling `include_metadata` omits offset, timestamp, stream, topic and partition
+columns. Checksum and origin timestamp flags work independently. The `id`,
+`payload` and `created_at` columns are always created. Existing tables are not
+migrated. With automatic creation disabled, startup does not check that a
+compatible target table exists.
+
+`id` is the full unsigned 128-bit message ID, without source identifiers or
+offset. Offset/checksum are cast to signed 64-bit integers and partition IDs to
+signed 32-bit integers, so values beyond those signed ranges appear negative.
+Message and origin timestamps are Unix microseconds; zero becomes the epoch and
+values outside the date library's range fall back to the current time.
+`created_at` is generated by PostgreSQL during insertion.
+
## Performance
### Recommended Indexes
+Create only indexes needed by your queries. Metadata indexes require the
+corresponding columns; the GIN index requires a JSONB payload column. The sink
+automatically creates only the primary-key index.
+
```sql
CREATE INDEX idx_iggy_messages_stream ON iggy_messages (iggy_stream);
CREATE INDEX idx_iggy_messages_topic ON iggy_messages (iggy_topic);
@@ -143,9 +187,17 @@ CREATE INDEX idx_payload_gin ON iggy_messages USING GIN (payload);
- Adjust `max_connections` based on PostgreSQL's `max_connections` setting
- Use `poll_interval` to control how often the sink checks for new messages
- Use `payload_format = "json"` for JSON data to enable native querying
+- Stay within PostgreSQL's 65,535 bind-parameter limit: each row uses
+ `2 + 5 * include_metadata + include_checksum + include_origin_timestamp`
+ parameters. With default flags, at most 7,281 rows fit in one statement.
+ Larger actual chunks fail; the runtime's `batch_length` also limits poll size.
## Example Configs
+These fragments replace the `streams` and `plugin_config` sections in the
+complete connector file above. Provision the named Iggy streams/topics and
+PostgreSQL databases, and replace the URI credentials before starting.
+
### JSON Messages with JSONB Storage
```toml
@@ -209,7 +261,28 @@ payload_format = "text"
### Automatic Retries
-The connector automatically retries transient database errors (connection issues, deadlocks, serialization failures) with exponential backoff. Configure with `max_retries` (default: 3) and `retry_delay` (default: `1s`). The actual delay is `retry_delay * attempt_number`. Non-transient errors fail immediately.
+The connector retries SQLx I/O and pool-acquisition timeout errors, plus SQLSTATEs
+`40001`, `40P01`, `57P01`, `57P02`, `57P03`, `08000`, `08003` and `08006`.
+`max_retries` counts total attempts, including the first (default: 3); zero still
+allows one attempt. Backoff is linear: `retry_delay * attempt_number`, with a
+one-based retry number. The base defaults to `1s`, including for invalid delay
+strings. Other errors stop that chunk's retry loop. Startup has no plugin retry
+loop.
+
+### Failed Inserts and Replay
+
+Each current-poll chunk becomes one multi-row `INSERT`. A bad payload, duplicate
+primary key, incompatible table or other terminal failure rejects that chunk.
+Later chunks are still attempted. Failures are logged and added to a private
+insertion-error count, but `consume()` returns success and counts every attempted
+message as processed. Runtime processed counts can include unstored messages,
+and runtime error counts do not expose these insert failures.
+
+The query has no `ON CONFLICT` or upsert clause. An ID already in the table
+rejects the whole chunk, including new IDs alongside it. There is no transaction
+covering a complete poll. Runtime auto-commits while polling and does not replay
+failed chunks. A connection failure can leave the write outcome uncertain;
+these retries do not establish an end-to-end at-least-once guarantee.
### Connection Pool Management
@@ -225,4 +298,10 @@ The sink can work with the source connector for pass-through scenarios:
For JSON data:
1. **Sink** with `payload_format = "json"` stores as JSONB
-2. **Source** with `payload_column = "data"` and `payload_format = "json_direct"` reads JSONB directly
+2. **Source** with `payload_column = "payload"` and `payload_format = "json_direct"` reads JSONB directly
+
+These settings transfer payloads. The source generates new message IDs and
+timestamps; it does not restore the sink's original message metadata. JSONB
+serialization does not preserve the original JSON bytes. See the
+[source README](../../sources/postgres_source/README.md) for the remaining
+required source configuration.
diff --git a/core/connectors/sinks/postgres_sink/src/lib.rs b/core/connectors/sinks/postgres_sink/src/lib.rs
index c3d89c36a..9f0ac08f0 100644
--- a/core/connectors/sinks/postgres_sink/src/lib.rs
+++ b/core/connectors/sinks/postgres_sink/src/lib.rs
@@ -231,7 +231,7 @@ impl PostgresSink {
messages: &[ConsumedMessage],
) -> Result<(), Error> {
let pool = self.get_pool()?;
- let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+ let batch_size = self.config.batch_size.unwrap_or(100).max(1) as usize;
for batch in messages.chunks(batch_size) {
if let Err(e) = self
@@ -562,6 +562,35 @@ mod tests {
}
}
+ #[test]
+ fn given_zero_batch_size_should_consume_empty_batch() {
+ let runtime = tokio::runtime::Runtime::new().expect("Failed to create test runtime");
+ runtime.block_on(async {
+ let mut config = test_config();
+ config.batch_size = Some(0);
+ let pool = PgPoolOptions::new()
+ .connect_lazy(config.connection_string.expose_secret())
+ .expect("Failed to configure lazy test pool");
+ let mut sink = PostgresSink::new(1, config);
+ sink.pool = Some(pool);
+
+ sink.consume(
+ &TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ },
+ MessagesMetadata {
+ partition_id: 0,
+ current_offset: 0,
+ schema: iggy_connector_sdk::Schema::Raw,
+ },
+ Vec::new(),
+ )
+ .await
+ .expect("An empty batch must succeed with a zero batch-size setting");
+ });
+ }
+
#[test]
fn given_json_format_should_return_json() {
assert_eq!(
diff --git a/core/connectors/sinks/quickwit_sink/README.md b/core/connectors/sinks/quickwit_sink/README.md
index f3067418a..f6f61febb 100644
--- a/core/connectors/sinks/quickwit_sink/README.md
+++ b/core/connectors/sinks/quickwit_sink/README.md
@@ -9,16 +9,16 @@ The Quickwit connector sends data to the Quickwit API using HTTP. It checks read
| `url` | Required | Quickwit base URL with an `http` or `https` scheme and host. Path prefixes and a trailing slash are supported; query strings and fragments are rejected. |
| `index` | Required | Index configuration as YAML, with a nonempty `index_id`. See the [Quickwit index configuration docs](https://quickwit.io/docs/configuration/index-config). |
| `verbose_logging` | `false` | Log received and ingested message counts at `info` instead of `debug`. |
-| `max_retries` | `3` | Total HTTP attempts including the first; `1` disables retries. |
+| `max_retries` | `3` | Total HTTP attempts including the first; `0` and `1` both allow one attempt. |
| `retry_delay` | `"1s"` | Base exponential delay for HTTP retries and readiness probes. |
-| `retry_max_delay` | `"5s"` | Maximum delay between HTTP retry attempts. |
-| `max_open_retries` | `10` | Total readiness probes including the first; `1` disables retries. |
+| `retry_max_delay` | `"5s"` | Cap for calculated HTTP retry delays; a valid `Retry-After` on HTTP 429 overrides it. |
+| `max_open_retries` | `10` | Total attempts per readiness check including the first; `0` and `1` both allow one attempt. |
| `open_retry_max_delay` | `"30s"` | Maximum delay between readiness probes. |
-| `timeout` | `"30s"` | Timeout for each HTTP request. |
+| `timeout` | `"30s"` | Timeout per HTTP attempt; retries and waits can extend the complete operation. |
Duration values require units, such as `250ms` or `30s`. Invalid or zero durations prevent initialization. Unknown plugin configuration keys are rejected.
-Set `plugin_config_format` in the connector TOML or with the `IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG_FORMAT` environment variable.
+The runtime's `plugin_config_format` can be set in the connector TOML or with the `IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG_FORMAT` environment variable. The `index` field itself is always a YAML string. The following fragment belongs in a complete sink entry with a plugin path and stream configuration; see the [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink/).
```toml
[plugin_config]
@@ -28,7 +28,7 @@ verbose_logging = false
max_retries = 3
retry_delay = "1s"
retry_max_delay = "5s"
-# Total readiness probes including the first; 1 disables retries.
+# Total attempts per readiness check including the first; 1 disables retries.
max_open_retries = 10
open_retry_max_delay = "30s"
timeout = "30s"
@@ -103,8 +103,12 @@ The examples use `mode: dynamic` to retain wrapper fields. With `mode: strict`,
## Delivery semantics
-Transient HTTP failures, including 429, can retry a request that Quickwit already accepted. Quickwit ingest has no deduplication key, so these retries can produce duplicate documents. Set `max_retries = 1` to disable HTTP retries and use at-most-once request submission.
+Transient HTTP failures, including 429, can retry a request that Quickwit already accepted. Quickwit ingest has no deduplication key, so these retries can produce duplicate documents. Set `max_retries = 1` to disable the sink's HTTP retry loop. Calculated delays use exponential backoff with jitter; a valid `Retry-After` on HTTP 429 replaces the calculated delay.
-The sink cannot guarantee at-least-once delivery. The runtime commits offsets when polling and ignores the plugin's consume return code. A permanent error or exhausted retry budget is logged, but affected messages are not redelivered. The runtime's processed-message count does not prove successful indexing. These runtime limitations are tracked in [#2927](https://github.com/apache/iggy/issues/2927) and [#2928](https://github.com/apache/iggy/issues/2928).
+Service readiness retries any failed health probe. After verifying or creating the index, the sink probes its ingest endpoint with an empty body before accepting messages.
+Index readiness retries HTTP 404, 429, 5xx and network failures.
+Each readiness check uses `max_open_retries` and `open_retry_max_delay`; these probes submit no documents.
-Chunks are independent: successful writes remain committed if another chunk fails. The sink continues later chunks and returns the last error. A successful ingest response acknowledges submission for indexing, not that every document passed the index mapping. This sink has no circuit breaker.
+The sink cannot guarantee at-least-once delivery. The runtime commits offsets when polling, logs/counts plugin callback errors and continues without replaying the failed batch. A permanent error or exhausted retry budget is logged, but affected messages are not redelivered. The runtime's processed-message count does not prove successful indexing.
+
+Chunks are independent: successful writes remain committed if another chunk fails. The sink continues later chunks and returns the last error. A successful ingest response acknowledges submission for indexing, not search visibility or acceptance of every document. The sink checks the HTTP status without inspecting per-document rejection counts. It does not add Iggy metadata or headers, and it does not update or compare an existing index's mapping. This sink has no circuit breaker.
diff --git a/core/connectors/sinks/quickwit_sink/config.toml b/core/connectors/sinks/quickwit_sink/config.toml
index c0f1a68b0..b947e2db6 100644
--- a/core/connectors/sinks/quickwit_sink/config.toml
+++ b/core/connectors/sinks/quickwit_sink/config.toml
@@ -39,7 +39,7 @@ verbose_logging = true
max_retries = 2
retry_delay = "250ms"
retry_max_delay = "2s"
-# Total readiness probes including the first; 1 disables retries.
+# Total attempts per readiness check including the first; 1 disables retries.
max_open_retries = 8
open_retry_max_delay = "10s"
timeout = "10s"
diff --git a/core/connectors/sinks/quickwit_sink/src/lib.rs b/core/connectors/sinks/quickwit_sink/src/lib.rs
index fadcb8b55..8bc7938c4 100644
--- a/core/connectors/sinks/quickwit_sink/src/lib.rs
+++ b/core/connectors/sinks/quickwit_sink/src/lib.rs
@@ -21,6 +21,7 @@ use async_trait::async_trait;
use base64::{Engine as _, engine::general_purpose};
use iggy_connector_sdk::retry::{
RetryPolicy, build_retry_client, check_connectivity_with_retry, is_transient_status,
+ retry_async,
};
use iggy_connector_sdk::{
ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
@@ -72,7 +73,7 @@ pub struct QuickwitSinkConfig {
pub retry_delay: Option<String>,
/// Maximum retry delay cap as a human-readable duration string, e.g. "5s" (default: 5s).
pub retry_max_delay: Option<String>,
- /// Total readiness attempts including the first (default: 10). 1 disables retries.
+ /// Total attempts per readiness check including the first (default: 10). 1 disables retries.
pub max_open_retries: Option<u32>,
/// Maximum retry delay cap when opening the sink, e.g. "30s" (default: 30s).
pub open_retry_max_delay: Option<String>,
@@ -172,6 +173,40 @@ impl QuickwitSink {
}
}
+ async fn wait_for_index(
+ &self,
+ client: &reqwest::Client,
+ policy: RetryPolicy,
+ ) -> Result<(), Error> {
+ // Index metadata can exist before Quickwit's ingest queue accepts documents.
+ retry_async(
+ policy,
+ &format!("Quickwit sink ID {} ingest readiness", self.id),
+ |error: &reqwest::Error| {
+ error.status().is_none_or(|status| {
+ status == StatusCode::NOT_FOUND || is_transient_status(status)
+ })
+ },
+ || async {
+ client
+ .post(&self.ingest_url)
+ .header(reqwest::header::CONTENT_LENGTH, "0")
+ .body("")
+ .send()
+ .await?
+ .error_for_status()
+ .map(|_| ())
+ },
+ )
+ .await
+ .map_err(|failure| {
+ Error::InitError(format!(
+ "Quickwit index '{}' ingest readiness: {failure}",
+ self.index_id
+ ))
+ })
+ }
+
async fn ingest(&self, messages: Vec<OwnedValue>) -> Result<(), Error> {
let capacity = messages
.len()
@@ -369,25 +404,26 @@ impl Sink for QuickwitSink {
.timeout(timeout)
.build()
.map_err(|error| Error::InitError(format!("reqwest client: {error}")))?;
+ let open_retry_policy = RetryPolicy {
+ max_attempts: self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES)
+ .max(1),
+ base_delay: retry_delay,
+ max_delay: open_retry_max_delay,
+ };
check_connectivity_with_retry(
&raw_client,
endpoint_url(&base_url, &["health", "readyz"])?,
"Quickwit sink",
self.id,
- RetryPolicy {
- max_attempts: self
- .config
- .max_open_retries
- .unwrap_or(DEFAULT_MAX_OPEN_RETRIES)
- .max(1),
- base_delay: retry_delay,
- max_delay: open_retry_max_delay,
- },
+ open_retry_policy,
)
.await?;
self.client = Some(build_retry_client(
- raw_client,
+ raw_client.clone(),
self.config
.max_retries
.unwrap_or(DEFAULT_MAX_RETRIES)
@@ -399,7 +435,7 @@ impl Sink for QuickwitSink {
if !self.has_index().await? {
self.create_index().await?;
}
- Ok(())
+ self.wait_for_index(&raw_client, open_retry_policy).await
}
.await;
if let Err(error) = result {
@@ -683,6 +719,11 @@ mod tests {
"index `test` already exist(s)",
),
("GET /prefix/api/v1/indexes/test HTTP/1.1", 200, "{}"),
+ (
+ "POST /prefix/api/v1/test/ingest?commit=auto HTTP/1.1",
+ 200,
+ "{}",
+ ),
])
.await;
let mut config = test_config();
@@ -726,14 +767,112 @@ mod tests {
});
}
+ #[test]
+ fn given_delayed_ingest_readiness_when_opened_should_deliver_the_first_batch() {
+ let runtime = test_runtime();
+ runtime.block_on(async {
+ for index_exists in [false, true] {
+ let mut responses = vec![
+ ("GET /health/readyz HTTP/1.1", 200, ""),
+ (
+ "GET /api/v1/indexes/test HTTP/1.1",
+ if index_exists { 200 } else { 404 },
+ "{}",
+ ),
+ ];
+ if !index_exists {
+ responses.push(("POST /api/v1/indexes HTTP/1.1", 200, "{}"));
+ }
+ let readiness_start = responses.len();
+ responses.extend([
+ (
+ "POST /api/v1/test/ingest?commit=auto HTTP/1.1",
+ 404,
+ "index not found",
+ ),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
+ ]);
+ let (url, server) = start_test_server(responses).await;
+ let mut config = test_config();
+ config.url = url;
+ config.max_open_retries = Some(3);
+ let mut sink = QuickwitSink::new(1, config);
+ sink.open()
+ .await
+ .expect("sink should wait for its ingest queue");
+ consume_payloads(&sink, vec![Payload::Text("first message".to_string())])
+ .await
+ .expect("first batch should reach the ready index");
+ let requests = server.await.expect("test server should finish");
+ assert!(
+ requests[readiness_start..requests.len() - 1]
+ .iter()
+ .all(Vec::is_empty)
+ );
+ let mut body = requests
+ .into_iter()
+ .last()
+ .expect("data request should exist");
+ assert_eq!(
+ simd_json::from_slice::<OwnedValue>(&mut body)
+ .expect("data request should be JSON"),
+ simd_json::json!({"text": "first message", "data_type": "text"})
+ );
+ }
+ });
+ }
+
+ #[test]
+ fn given_unavailable_ingest_readiness_when_opened_should_stop_within_the_attempt_budget() {
+ let runtime = test_runtime();
+ runtime.block_on(async {
+ for (status, max_attempts) in
+ [(404, 0), (404, 1), (404, 3), (429, 3), (503, 3), (401, 3)]
+ {
+ let expected_attempts = if status == 401 {
+ 1
+ } else {
+ max_attempts.max(1)
+ };
+ let mut responses = vec![
+ ("GET /health/readyz HTTP/1.1", 200, ""),
+ ("GET /api/v1/indexes/test HTTP/1.1", 200, "{}"),
+ ];
+ responses.extend(std::iter::repeat_n(
+ (
+ "POST /api/v1/test/ingest?commit=auto HTTP/1.1",
+ status,
+ "not ready",
+ ),
+ expected_attempts as usize,
+ ));
+ let (url, server) = start_test_server(responses).await;
+ let mut config = test_config();
+ config.url = url;
+ config.max_open_retries = Some(max_attempts);
+ let mut sink = QuickwitSink::new(1, config);
+ assert!(
+ matches!(sink.open().await, Err(Error::InitError(_))),
+ "status {status}"
+ );
+ assert!(sink.client.is_none());
+ let requests = server.await.expect("test server should finish");
+ assert_eq!(requests.len(), expected_attempts as usize + 2);
+ assert!(requests.iter().all(Vec::is_empty));
+ }
+ });
+ }
+
#[test]
fn given_ingest_failure_when_consumed_should_classify_http_errors() {
let runtime = test_runtime();
runtime.block_on(async {
- for status in [400, 429, 503] {
+ for status in [400, 404, 429, 503] {
let (url, server) = start_test_server(vec![
("GET /health/readyz HTTP/1.1", 200, ""),
("GET /api/v1/indexes/test HTTP/1.1", 200, "{}"),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
(
"POST /api/v1/test/ingest?commit=auto HTTP/1.1",
status,
@@ -747,7 +886,7 @@ mod tests {
sink.open().await.expect("sink should open");
let result =
consume_payloads(&sink, vec![Payload::Text("message".to_string())]).await;
- if status == 400 {
+ if status == 400 || status == 404 {
assert!(matches!(result, Err(Error::PermanentHttpError(_))));
} else {
assert!(matches!(result, Err(Error::HttpRequestFailed(_))));
@@ -773,6 +912,7 @@ mod tests {
let (url, server) = start_test_server(vec![
("GET /health/readyz HTTP/1.1", 200, ""),
("GET /api/v1/indexes/test HTTP/1.1", 200, "{}"),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
(
"POST /api/v1/test/ingest?commit=auto HTTP/1.1",
503,
@@ -802,8 +942,8 @@ mod tests {
.await;
assert!(matches!(result, Err(Error::HttpRequestFailed(_))));
let requests = server.await.expect("test server should finish");
- assert_eq!(requests[2].len(), MAX_INGEST_BODY_BYTES);
- assert_eq!(requests[2].last(), Some(&b'\n'));
+ assert_eq!(requests[3].len(), MAX_INGEST_BODY_BYTES);
+ assert_eq!(requests[3].last(), Some(&b'\n'));
let mut last_body = requests
.into_iter()
.last()
@@ -824,6 +964,7 @@ mod tests {
("GET /health/readyz HTTP/1.1", 200, ""),
("GET /api/v1/indexes/test HTTP/1.1", 200, "{}"),
("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
+ ("POST /api/v1/test/ingest?commit=auto HTTP/1.1", 200, "{}"),
])
.await;
let mut config = test_config();
@@ -841,7 +982,7 @@ mod tests {
.await;
assert!(matches!(result, Err(Error::InvalidRecordValue(_))));
let requests = server.await.expect("test server should finish");
- assert_eq!(requests[2], b"{\"sequence\":1}\n{\"sequence\":3}\n");
+ assert_eq!(requests[3], b"{\"sequence\":1}\n{\"sequence\":3}\n");
});
}
@@ -926,18 +1067,20 @@ mod tests {
.next()
.expect("request line should exist")
.to_string();
- let content_length = headers
- .lines()
- .find_map(|line| {
- let (name, value) = line.split_once(':')?;
- name.eq_ignore_ascii_case("content-length").then(|| {
- value
- .trim()
- .parse::<usize>()
- .expect("content length should be valid")
- })
+ let content_length = headers.lines().find_map(|line| {
+ let (name, value) = line.split_once(':')?;
+ name.eq_ignore_ascii_case("content-length").then(|| {
+ value
+ .trim()
+ .parse::<usize>()
+ .expect("content length should be valid")
})
- .unwrap_or(0);
+ });
+ assert!(
+ !request_line.starts_with("POST ") || content_length.is_some(),
+ "Quickwit requires an explicit Content-Length for POST requests"
+ );
+ let content_length = content_length.unwrap_or(0);
let body_start = headers_end + b"\r\n\r\n".len();
while buffer.len() < body_start + content_length {
let read = stream.read(&mut chunk).await.expect("body should read");
diff --git a/core/connectors/sinks/s3_sink/Cargo.toml b/core/connectors/sinks/s3_sink/Cargo.toml
index 4ca39a179..0e72d6867 100644
--- a/core/connectors/sinks/s3_sink/Cargo.toml
+++ b/core/connectors/sinks/s3_sink/Cargo.toml
@@ -56,6 +56,7 @@ tracing = { workspace = true }
[dev-dependencies]
simd-json = { workspace = true }
+wiremock = { workspace = true }
[lints]
workspace = true
diff --git a/core/connectors/sinks/s3_sink/README.md b/core/connectors/sinks/s3_sink/README.md
index 6ee55368c..bb25c7bd4 100644
--- a/core/connectors/sinks/s3_sink/README.md
+++ b/core/connectors/sinks/s3_sink/README.md
@@ -16,14 +16,17 @@ Writes messages from Iggy streams to Amazon S3 and S3-compatible object stores (
### Connector Runtime Config
+Build and start the matching 0.9.0/edge broker and connector runtime from the checkout root. Use the [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink/) for broker credentials and the main runtime configuration. Save the following connector entry in its configured connector directory and append one `[plugin_config]` table. Create the bucket separately before startup.
+
```toml
type = "sink"
key = "s3"
enabled = true
version = 0
name = "S3 sink"
-path = "../../target/release/libiggy_connector_s3_sink"
+path = "target/release/libiggy_connector_s3_sink"
verbose = false
+plugin_config_format = "toml"
[[streams]]
stream = "application_logs"
@@ -60,21 +63,25 @@ retry_delay = "1s"
| ------ | ---- | ------- | ----------- |
| `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 |
+| `prefix` | String | `None` | Prefix for data objects and loss markers; outer slashes are removed |
| `endpoint` | String | `None` | Custom S3 endpoint for MinIO, R2, etc. |
| `access_key_id` | String | `None` | AWS access key; omit for env/instance profile |
| `secret_access_key` | String | `None` | AWS secret key; omit for env/instance profile |
| `path_template` | String | `{stream}/{topic}/{date}/{hour}` | Template for S3 key directory structure |
| `file_rotation` | String | `size` | Rotation strategy: `size` or `messages` |
-| `max_file_size` | String | `8MiB` | Max file size before rotation (when `file_rotation = "size"`) |
-| `max_messages_per_file` | Integer | `None` | Max messages per file (required when `file_rotation = "messages"`) |
+| `max_file_size` | String | `8MiB` | Size rotation threshold, positive and at most `5GiB`; validated in both modes |
+| `max_messages_per_file` | Integer | `None` | Positive count required in `messages` mode; ignored by size rotation |
| `output_format` | String | `json_lines` | Output format: `json_lines`, `json_array`, or `raw` |
-| `include_metadata` | Boolean | `true` | Include stream/topic/partition/offset in output |
-| `include_headers` | Boolean | `false` | Include message headers in output |
-| `max_attempts` | Integer | `3` | Max total upload attempts per file (also accepts `max_retries` as alias) |
-| `retry_delay` | String | `1s` | Base delay for exponential backoff (humantime format) |
+| `include_metadata` | Boolean | `true` | Include stream/topic/partition/offset/timestamp in JSON output |
+| `include_headers` | Boolean | `false` | Include available headers in JSON output, independently of metadata |
+| `max_attempts` | Integer | `3` | Outer upload attempts; `0` and `1` both allow one (alias: `max_retries`) |
+| `retry_delay` | String | `1s` | Base exponential delay (humantime format); zero allowed, invalid values reject startup |
| `path_style` | Boolean | auto | Force path-style S3 addressing; auto-enabled when `endpoint` is set |
+Each stream/topic/partition has its own buffer across polls. Rotation checks run after appending a record. Size mode counts formatted record bytes without JSON delimiters/brackets, so the threshold is not a hard object-size limit. Message-count mode does not also enforce the size threshold. There is no timer flush: partial buffers wait for the threshold or close. Uploads use one `PutObject`, not multipart upload, and must fit the destination's single-upload limit.
+
+Format names are case-insensitive; `jsonl` and `jsonlines` alias `json_lines`.
+
### Path Template Variables
| Variable | Description | Example |
@@ -88,22 +95,30 @@ retry_delay = "1s"
**Note:** `{timestamp}`, `{date}`, and `{hour}` are all derived from the first message timestamp in each buffer. They are deterministic across retries within the same process. However, a process restart resets in-memory buffers, so batch boundaries (and therefore timestamps in the key) may differ after recovery.
+Stream/topic substitutions keep ASCII letters, digits, `.`, `_` and `-`, replacing other characters with `_`. Ensure names remain distinct after sanitization and keep stream/topic separation in the template. Different inputs can otherwise overwrite the same object key.
+The filename includes partition ID (at least five digits) and first/last offsets (twenty digits each), with `.jsonl`, `.json` or `.bin` extension. Bucket versioning can retain overwritten versions; offset naming is not a global deduplication guarantee.
+
### Credentials
-Credentials can be provided in three ways (in order of precedence):
+The pinned `aws-creds` chain tries these sources in order:
+
+1. **Explicit config**: both `access_key_id` and `secret_access_key`, or neither. This path has no session-token option.
+2. **Environment**: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, optional `AWS_SESSION_TOKEN` / `AWS_SECURITY_TOKEN`.
+3. **Credentials file**: `AWS_SHARED_CREDENTIALS_FILE` or `~/.aws/credentials`, using `[default]`. The plugin does not select `AWS_PROFILE`.
+4. **STS web identity**: `AWS_ROLE_ARN` and `AWS_WEB_IDENTITY_TOKEN_FILE`.
+5. **ECS relative-URI credentials**: `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI`.
+6. **EC2 instance metadata**: IMDSv2, then v1.
-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
+For temporary key pairs, supply the token through the environment or shared credentials file. This is not the AWS SDK credential chain.
-Both `access_key_id` and `secret_access_key` must be provided together or both omitted.
+Startup writes an empty object at bucket-root `.iggy-sink-probe`, ignoring `prefix`, and then attempts deletion. It requires `PutObject` on the probe key, not `ListBucket`; non-success HTTP statuses prevent startup. Delete permission is optional and cleanup errors are ignored. Reserve this key: any existing object there is overwritten and, if cleanup succeeds, deleted.
## 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
+s3://my-data-lake/iggy/raw/application_logs/api_requests/2026-03-16/14/00001-00000000000000000000-00000000000000000999.jsonl
```
Each line:
@@ -112,8 +127,15 @@ Each line:
{"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}}
```
+Both JSON formats wrap the payload under `payload`, including when metadata is disabled. JSON values retain their shape; text and Proto variants become strings. Valid JSON raw bytes are parsed; other raw bytes, Avro and FlatBuffer variants become base64 strings without an encoding tag.
+Metadata timestamps use UTC second precision; IDs, checksums and origin timestamps are omitted. Headers are independent of metadata: strings remain strings, raw values become base64, booleans remain booleans, supported numeric values become JSON numbers, and other values use strings.
+
+Raw output concatenates payload bytes without delimiters and ignores both inclusion flags. Use `schema = "raw"` without payload-changing transforms to preserve original bytes; JSON decoding followed by raw output can reserialize the payload.
+
## S3-Compatible Stores
+Replace the earlier `[plugin_config]` table with one of these examples. Supply an existing bucket and credentials with the required permissions. The [website walkthrough](https://iggy.apache.org/docs/connectors/sinks/s3/) includes local MinIO setup and Iggy CLI commands.
+
### MinIO
```toml
@@ -138,22 +160,19 @@ secret_access_key = "..."
## Delivery Semantics
-All retry logic lives inside `consume()`. The connector runtime invokes `consume()` via an FFI callback that returns an `i32` status code. The runtime does not inspect this return value (see `process_messages()` in `runtime/src/sink.rs`), so errors logged by the sink are not propagated to the runtime's retry or alerting mechanisms. Additionally, consumer group offsets are committed before processing ([runtime issue #1](#known-limitations)). This means:
-
-- Failed messages are **not retried by the runtime** — only by the sink's internal retry loop
-- Messages are committed **before delivery** — a crash after commit but before delivery loses messages
+The runtime auto-commits while polling, logs/counts a failed plugin callback, and continues without replaying that batch. A successful callback may only mean that messages were buffered; processed counts do not establish S3 delivery. A crash loses unflushed buffers.
-The effective delivery guarantee is **at-most-once** at the runtime level. The sink's internal retries provide best-effort delivery within each `consume()` call.
+The sink retries HTTP 408, 429, 5xx and client errors, using exponential backoff with jitter capped at 60 seconds. Other HTTP statuses stop that upload. `Retry-After` is not parsed. The S3 library also retries transport errors once internally and has a 60-second request timeout, so `max_attempts` counts outer `PutObject` calls, not individual network requests or a total time budget. The startup probe does not use the sink's upload retry loop.
## Known Limitations
-1. **Runtime ignores `consume()` status**: The connector runtime invokes `consume()` via an FFI callback returning `i32`. The `process_messages()` function in `runtime/src/sink.rs` does not inspect the return value. Errors are logged internally by the sink but do not trigger runtime-level retry or alerting. ([#2927](https://github.com/apache/iggy/issues/2927))
+1. **No runtime replay**: Auto-commit happens while polling, before the plugin finishes delivery. Callback failures are counted, but the runtime continues polling. There is no end-to-end delivery or deduplication guarantee.
-2. **Offsets committed before processing**: The `PollingMessages` auto-commit strategy commits consumer group offsets before `consume()` is called. Combined with limitation 1, at-least-once delivery is not achievable. ([#2928](https://github.com/apache/iggy/issues/2928))
+2. **In-memory buffering only**: There is no write-ahead log or timer flush. A crash loses unflushed buffers. Buffers are cleared before upload; a failed upload loses that file's messages and skips the remainder of the current poll.
-3. **In-memory buffering only**: There is no write-ahead log. A process crash loses all in-memory buffered messages that have not yet been flushed to S3.
+3. **Loss markers are best effort**: After an upload failure the sink tries to write `<object-key>.lost` with offset range, message count and error. The marker has no message payload, is not a dead-letter queue, and can also fail. Its upload uses the same retry policy.
-4. **No dead letter queue**: Failed messages are logged at `error!` level but not persisted to a DLQ. DLQ support would be a runtime-level feature.
+4. **Close does not report flush failures**: Graceful close attempts remaining uploads, logs errors and returns success. Shutdown success does not establish that every message was stored.
## Building
@@ -161,4 +180,4 @@ The effective delivery guarantee is **at-most-once** at the runtime level. The s
cargo build --release -p iggy_connector_s3_sink
```
-The compiled plugin will be at `target/release/libiggy_connector_s3_sink.{so,dylib,dll}`.
+The Linux plugin is `target/release/libiggy_connector_s3_sink.so`. The runtime configuration uses the extensionless path shown above.
diff --git a/core/connectors/sinks/s3_sink/config.toml b/core/connectors/sinks/s3_sink/config.toml
index ed71fc900..b53f492b4 100644
--- a/core/connectors/sinks/s3_sink/config.toml
+++ b/core/connectors/sinks/s3_sink/config.toml
@@ -20,7 +20,7 @@ key = "s3"
enabled = true
version = 0
name = "S3 sink"
-path = "../../target/release/libiggy_connector_s3_sink"
+path = "target/release/libiggy_connector_s3_sink"
verbose = false
[[streams]]
diff --git a/core/connectors/sinks/s3_sink/src/client.rs b/core/connectors/sinks/s3_sink/src/client.rs
index 3ec489f1b..df14e80ad 100644
--- a/core/connectors/sinks/s3_sink/src/client.rs
+++ b/core/connectors/sinks/s3_sink/src/client.rs
@@ -83,18 +83,28 @@ pub(crate) async fn create_bucket(config: &S3SinkConfig) -> Result<Box<Bucket>,
/// and fails for write-only IAM policies.
pub(crate) async fn verify_bucket(bucket: &Bucket) -> Result<(), Error> {
const PROBE_KEY: &str = ".iggy-sink-probe";
- bucket.put_object(PROBE_KEY, &[]).await.map_err(|e| {
+ let response = bucket.put_object(PROBE_KEY, &[]).await.map_err(|e| {
Error::InitError(format!(
"S3 bucket '{}' connectivity check failed: {e}",
bucket.name
))
})?;
+ let status = response.status_code();
+ if !(200..300).contains(&status) {
+ return Err(Error::InitError(format!(
+ "S3 bucket '{}' connectivity check returned status {status}",
+ bucket.name
+ )));
+ }
let _ = bucket.delete_object(PROBE_KEY).await;
Ok(())
}
#[cfg(test)]
mod tests {
+ use wiremock::matchers::method;
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
use super::*;
use crate::{
FileRotation, default_max_file_size, default_output_format, default_path_template,
@@ -121,6 +131,36 @@ mod tests {
}
}
+ #[test]
+ fn given_probe_response_should_require_success_status() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for status in [200, 403, 404, 503] {
+ let server = MockServer::start().await;
+ Mock::given(method("PUT"))
+ .respond_with(ResponseTemplate::new(status))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ ..base_config()
+ };
+ let bucket = create_bucket(&config).await.expect("Create S3 test client");
+ let result = verify_bucket(&bucket).await;
+ if status == 200 {
+ assert!(result.is_ok(), "Successful probe must pass: {result:?}");
+ } else {
+ assert!(
+ matches!(result, Err(Error::InitError(_))),
+ "Failed probe status {status} must prevent initialization: {result:?}"
+ );
+ }
+ }
+ });
+ }
+
#[test]
fn validate_both_credentials_present() {
let config = S3SinkConfig {
diff --git a/core/connectors/sinks/s3_sink/src/sink.rs b/core/connectors/sinks/s3_sink/src/sink.rs
index 7ca3e1296..66c990d36 100644
--- a/core/connectors/sinks/s3_sink/src/sink.rs
+++ b/core/connectors/sinks/s3_sink/src/sink.rs
@@ -227,11 +227,11 @@ impl S3Sink {
}
};
+ *processed += 1;
+
if let Some(payload) = flush_payload {
self.do_upload(bucket, payload).await?;
}
-
- *processed += 1;
}
Ok(())
}
@@ -371,7 +371,7 @@ impl S3Sink {
attempt += 1;
if attempt >= max_attempts {
return Err(Error::CannotStoreData(format!(
- "S3 PutObject returned status {status} after {max_attempts} attempts for key '{s3_key}'"
+ "S3 PutObject returned status {status} after {attempt} attempts for key '{s3_key}'"
)));
}
warn!(
@@ -383,7 +383,7 @@ impl S3Sink {
attempt += 1;
if attempt >= max_attempts {
return Err(Error::CannotStoreData(format!(
- "S3 PutObject failed after {max_attempts} attempts for key '{s3_key}': {e}"
+ "S3 PutObject failed after {attempt} attempts for key '{s3_key}': {e}"
)));
}
warn!(
@@ -404,6 +404,10 @@ fn is_retriable_status(status: u16) -> bool {
#[cfg(test)]
mod tests {
+ use iggy_connector_sdk::{Payload, Schema};
+ use wiremock::matchers::{method, path};
+ use wiremock::{Mock, MockServer, ResponseTemplate};
+
use super::*;
use crate::{
DEFAULT_MAX_FILE_SIZE, DEFAULT_OUTPUT_FORMAT, DEFAULT_PATH_TEMPLATE, FileRotation, S3Sink,
@@ -431,6 +435,110 @@ mod tests {
}
}
+ #[test]
+ fn given_failed_upload_should_count_each_lost_message_once() {
+ const TOTAL_MESSAGES: u64 = 3;
+ const ROTATION_MESSAGES: u64 = 2;
+
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ for previously_buffered in [0, 1] {
+ let server = MockServer::start().await;
+ Mock::given(method("PUT"))
+ .and(path("/test-bucket/.iggy-sink-probe"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ file_rotation: FileRotation::Messages,
+ max_messages_per_file: Some(ROTATION_MESSAGES),
+ max_attempts: Some(1),
+ output_format: "raw".to_string(),
+ ..test_config()
+ };
+ let mut sink = S3Sink::new(1, config);
+ sink.open().await.expect("Successful startup probe");
+ let topic = TopicMetadata {
+ stream: "events".to_string(),
+ topic: "messages".to_string(),
+ };
+ let metadata = || MessagesMetadata {
+ partition_id: 0,
+ current_offset: 0,
+ schema: Schema::Raw,
+ };
+ let message = |offset| ConsumedMessage {
+ id: u128::from(offset) + 1,
+ offset,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(vec![1]),
+ };
+ sink.consume(
+ &topic,
+ metadata(),
+ (0..previously_buffered).map(message).collect(),
+ )
+ .await
+ .expect("Initial messages must remain buffered");
+ let result = sink
+ .consume(
+ &topic,
+ metadata(),
+ (previously_buffered..TOTAL_MESSAGES).map(message).collect(),
+ )
+ .await;
+ assert!(matches!(result, Err(Error::CannotStoreData(_))));
+ let state = sink.state.lock().await;
+ assert_eq!(state.messages_received, TOTAL_MESSAGES);
+ assert_eq!(state.messages_uploaded, 0);
+ assert_eq!(
+ state.messages_lost, TOTAL_MESSAGES,
+ "Upload failure must count the flushed buffer and unprocessed tail once, with {previously_buffered} previously buffered messages"
+ );
+ }
+ });
+ }
+
+ #[test]
+ fn given_zero_attempts_should_report_one_failed_upload() {
+ let runtime = tokio::runtime::Runtime::new().expect("Start test runtime");
+ runtime.block_on(async {
+ let server = MockServer::start().await;
+ Mock::given(method("PUT"))
+ .respond_with(ResponseTemplate::new(503))
+ .expect(1)
+ .mount(&server)
+ .await;
+ let config = S3SinkConfig {
+ endpoint: Some(server.uri()),
+ access_key_id: Some("test-access-key".into()),
+ secret_access_key: Some("test-secret-key".into()),
+ max_attempts: Some(0),
+ ..test_config()
+ };
+ let bucket = crate::client::create_bucket(&config)
+ .await
+ .expect("Create test client");
+ let mut sink = S3Sink::new(1, config);
+ sink.validate_and_parse_config()
+ .expect("Valid configuration");
+ let error = sink
+ .upload_with_retry(&bucket, "message.bin", b"payload")
+ .await
+ .expect_err("The only upload must fail");
+ assert!(
+ error.to_string().contains("after 1 attempts"),
+ "Report the actual upload count: {error}"
+ );
+ });
+ }
+
#[test]
fn is_retriable_5xx() {
assert!(is_retriable_status(500));
diff --git a/core/connectors/sinks/surrealdb_sink/README.md b/core/connectors/sinks/surrealdb_sink/README.md
index ba2a5e039..9a8c731d0 100644
--- a/core/connectors/sinks/surrealdb_sink/README.md
+++ b/core/connectors/sinks/surrealdb_sink/README.md
@@ -2,24 +2,37 @@
Writes Apache Iggy stream messages into SurrealDB over the HTTP API.
-The sink writes one SurrealQL bulk `INSERT IGNORE` per connector batch. Each
-record uses a deterministic SurrealDB record id derived from stream, topic,
-partition, offset and Iggy message id, so replayed batches are idempotent and
-existing records are left untouched.
+The sink splits each poll into chunks of at most `batch_size` messages and writes
+one SurrealQL bulk `INSERT IGNORE ... RETURN NONE` per chunk containing valid
+records. Each record uses a deterministic SurrealDB record id derived from
+stream, topic, partition, offset and Iggy message id. Replaying those identities
+leaves existing records and their payloads untouched. There is no cross-poll buffer.
-Persistent sink failures are at-most-once from the runtime's perspective:
-messages may already be committed in Iggy before this connector exhausts its
-write attempts, so failed writes are logged but not redelivered.
+The runtime auto-commits while polling, logs/counts failed plugin callbacks, and
+continues without replaying them. Deterministic IDs protect repeated writes of
+the same identities, but do not ensure that a failed batch will be delivered.
## Configuration
+From the matching 0.9.0/edge Iggy checkout root, build the plugin:
+
+```bash
+cargo build --release -p iggy_connector_surrealdb_sink
+```
+
+Use the [sink guide](https://iggy.apache.org/docs/connectors/sinks/sink/) for the
+broker credentials and main runtime configuration. Save this connector entry in
+its connector directory and start the runtime from the checkout root. The
+[SurrealDB walkthrough](https://iggy.apache.org/docs/connectors/sinks/surrealdb/)
+includes local backend, Iggy CLI and SQL query commands.
+
```toml
type = "sink"
key = "surrealdb"
enabled = true
version = 0
name = "SurrealDB sink"
-path = "../../target/release/libiggy_connector_surrealdb_sink"
+path = "target/release/libiggy_connector_surrealdb_sink"
plugin_config_format = "toml"
[[streams]]
@@ -62,38 +75,84 @@ verbose_logging = false
| `namespace` | required | SurrealDB namespace selected during `open()`. |
| `database` | required | SurrealDB database selected during `open()`. |
| `table` | required | Target table. Must be a safe SurrealQL identifier. |
-| `username` / `password` | none | Optional credentials. |
+| `username` / `password` | none | Both required unless `auth_scope = "none"`. |
| `auth_scope` | `root` | `root`, `namespace`, `database`, or `none`. |
| `use_tls` | `false` | Uses `https://` when true and `endpoint` has no scheme, `http://` otherwise. |
-| `auto_define_table` | `false` | Runs `DEFINE TABLE IF NOT EXISTS <table> SCHEMALESS`. |
-| `define_indexes` | `false` | Defines an offset index on stream/topic/partition/offset. Requires `auto_define_table`. |
-| `batch_size` | `1000` | Maximum number of records per SurrealDB request. |
+| `auto_define_table` | `false` | Creates missing namespace/database and a schemaless table; requires root authentication scope. |
+| `define_indexes` | `false` | Defines a non-unique stream/topic/partition/offset index; requires metadata and only runs with automatic DDL. |
+| `batch_size` | `1000` | Maximum input records per insert chunk; zero is raised to one. |
| `payload_format` | `auto` | `auto`, `json`, `text`, `base64`, or `binary` (`binary` is an alias for `base64`). |
-| `include_metadata` | `true` | Stores stream/topic/partition/offset/timestamps/schema fields. |
+| `include_metadata` | `true` | Stores stream/topic/partition/offset/timestamp/schema fields. |
| `include_headers` | `true` | Stores Iggy headers as a deterministic object. Raw headers are base64 encoded. |
| `include_checksum` | `true` | Stores `iggy_checksum`. |
| `include_origin_timestamp` | `true` | Stores `iggy_origin_timestamp`. |
-| `query_timeout` | `30s` | SurrealDB HTTP request timeout. |
+| `query_timeout` | `30s` | Timeout per HTTP request, not a total retry/reconnect budget. |
| `max_retries` | `3` | Total attempts for transient write failures. Values below `1` are raised to `1`. |
| `retry_delay` | `100ms` | Base retry delay. |
| `max_retry_delay` | `5s` | Capped exponential retry delay. |
| `verbose_logging` | `false` | Emits per-batch success logs at `info`. |
+Namespace, database and table names must start with an ASCII letter or underscore
+and contain only ASCII letters, digits and underscores. Endpoints reject embedded
+credentials, paths beyond `/`, query strings and fragments. An explicit HTTP(S)
+scheme takes precedence over `use_tls`. Authentication-scope and payload-format
+names are case-insensitive; unknown values reject startup.
+
+Root/namespace/database authentication requires both credentials. Startup checks
+`/signin`; subsequent SQL uses Basic authentication with the configured scope.
+Namespace/database users require pre-existing resources and automatic DDL disabled.
+`auth_scope = "none"` ignores credentials and requires a server that permits the
+unauthenticated operations. Automatic DDL uses `IF NOT EXISTS` and does not
+migrate existing definitions. Index creation uses `<table>_iggy_offset_idx`.
+Disabling metadata with indexes enabled rejects startup; disabling automatic DDL
+with indexes enabled logs a warning and skips index creation.
+
+Invalid duration strings warn and fall back to `1s`; zero is accepted. A maximum
+retry delay smaller than the base delay is raised to the base delay.
+
## Stored Shape
-With metadata enabled, records contain:
+Every submitted record contains:
- `id`: deterministic SurrealDB record id key
- `iggy_message_id`: original Iggy message id as a string
-- `iggy_stream`, `iggy_topic`, `iggy_partition_id`, `iggy_offset`
-- `iggy_timestamp`, `iggy_origin_timestamp`, `iggy_checksum`, `iggy_schema`
-- `iggy_headers`
- `payload`
- `payload_encoding`
+`include_metadata` adds `iggy_stream`, `iggy_topic`, `iggy_partition_id`,
+`iggy_offset`, `iggy_timestamp` and `iggy_schema`. The checksum, origin timestamp
+and headers have independent inclusion flags. `iggy_headers` is omitted when
+headers are absent or empty. Non-raw headers are strings, including numeric and
+boolean values; raw headers use `{"data":"AQID","iggy_header_encoding":"base64"}`.
+
+Partition IDs, offsets, timestamps and checksums are strings. Timestamps retain
+Iggy's microsecond values. Record keys encode stream/topic UTF-8 bytes as hex,
+then partition, offset and the 32-digit hexadecimal message ID. Payload and
+transform changes do not alter that identity.
+
`payload_format = "auto"` stores decoded JSON payloads as queryable SurrealDB
-values, text payloads as strings, and binary payloads as base64 strings.
+values, text/Proto variants as strings, and raw/Avro/FlatBuffer bytes as base64
+strings, even when raw bytes contain valid JSON. Explicit `json` parses other
+payload variants as JSON, `text` requires UTF-8, and `base64`/`binary` encodes the
+payload bytes. Invalid conversions reject that record. Destination schema,
+numeric and other value constraints still apply.
+
+The sink's `messages_processed` counter counts records submitted in successful
+SQL statements, including ignored records. On SurrealDB 3.1.4, `INSERT IGNORE`
+also silently skips unique-index conflicts and field-assertion failures while
+accepting valid records from the same chunk. Runtime counters instead reflect
+whether the whole callback succeeded. Neither counts newly inserted rows.
+
+## Failures and Retries
+
+Malformed records are skipped while valid records in the same chunk are still
+submitted. A failed chunk does not stop later chunks; the last error is returned.
+HTTP status and every SQL statement status are checked. Successful chunks are not
+rolled back when another chunk fails.
-The `messages_processed` counter reports valid records submitted to SurrealDB.
-With `INSERT IGNORE`, duplicates can be ignored by SurrealDB while still being
-counted as submitted.
+The sink retries transaction-conflict errors, connection/timeouts and HTTP 408,
+429, 500, 502, 503 and 504, using capped exponential backoff with jitter.
+`Retry-After` is not used. Connection errors trigger reconnection and repeat
+sign-in, health and optional DDL; failed reconnection stops that chunk.
+Deterministic `INSERT IGNORE` IDs preserve existing records but provide no
+end-to-end at-least-once delivery guarantee.
diff --git a/core/connectors/sinks/surrealdb_sink/config.toml b/core/connectors/sinks/surrealdb_sink/config.toml
index dd37160c6..e7b0dc3fd 100644
--- a/core/connectors/sinks/surrealdb_sink/config.toml
+++ b/core/connectors/sinks/surrealdb_sink/config.toml
@@ -20,7 +20,7 @@ key = "surrealdb"
enabled = true
version = 0
name = "SurrealDB sink"
-path = "../../target/release/libiggy_connector_surrealdb_sink"
+path = "target/release/libiggy_connector_surrealdb_sink"
plugin_config_format = "toml"
verbose = false
benchmark = false
diff --git a/core/connectors/sinks/surrealdb_sink/src/lib.rs b/core/connectors/sinks/surrealdb_sink/src/lib.rs
index 0d774083f..b697ceda1 100644
--- a/core/connectors/sinks/surrealdb_sink/src/lib.rs
+++ b/core/connectors/sinks/surrealdb_sink/src/lib.rs
@@ -583,7 +583,14 @@ impl SurrealDbSink {
return request;
};
- request.basic_auth(username, Some(password.expose_secret()))
+ let mut request = request.basic_auth(username, Some(password.expose_secret()));
+ if matches!(self.auth_scope, AuthScope::Namespace | AuthScope::Database) {
+ request = request.header("Surreal-Auth-NS", &self.namespace);
+ }
+ if self.auth_scope == AuthScope::Database {
+ request = request.header("Surreal-Auth-DB", &self.database);
+ }
+ request
}
async fn process_messages(
diff --git a/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs b/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
index ae46c0a75..667aced9b 100644
--- a/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
+++ b/core/integration/tests/connectors/elasticsearch/elasticsearch_sink.rs
@@ -24,6 +24,11 @@ use iggy_common::Identifier;
use iggy_common::MessageClient;
use integration::harness::seeds;
use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::{sleep, timeout};
+
+const LOG_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
+const LOG_POLL_INTERVAL: Duration = Duration::from_millis(100);
#[iggy_harness(
server(connectors_runtime(config_path = "tests/connectors/elasticsearch/sink.toml")),
@@ -214,3 +219,55 @@ async fn elasticsearch_sink_preserves_json_structure(
complex_messages.len()
);
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/elasticsearch/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_rejected_document_when_indexing_should_report_zero_indexed(
+ harness: &TestHarness,
+ fixture: ElasticsearchSinkFixture,
+) {
+ let client = harness.root_client().await.expect("root client");
+ let stream_id: Identifier = seeds::names::STREAM.try_into().expect("stream ID");
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().expect("topic ID");
+ let mut message = IggyMessage::builder()
+ .id(1)
+ .payload(Bytes::from_static(b"[1,2]"))
+ .build()
+ .expect("array payload message");
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ std::slice::from_mut(&mut message),
+ )
+ .await
+ .expect("send rejected document");
+
+ let runtime = harness.connectors_runtime().expect("connectors runtime");
+ let logs = timeout(LOG_WAIT_TIMEOUT, async {
+ loop {
+ let (stdout, stderr) = runtime.collect_logs();
+ let logs = format!("{stdout}\n{stderr}");
+ if logs.contains("Successfully indexed") {
+ break logs;
+ }
+ sleep(LOG_POLL_INTERVAL).await;
+ }
+ })
+ .await
+ .expect("bulk completion should be logged");
+
+ assert!(logs.contains("Document indexing error:"), "{logs}");
+ assert_eq!(
+ fixture.get_document_count().await.expect("document count"),
+ 0,
+ "Elasticsearch must reject the non-object document"
+ );
+ assert!(
+ logs.contains("Successfully indexed 0 documents"),
+ "the indexed count must exclude the rejected document: {logs}"
+ );
+}
diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
index f7af06144..0cba20394 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -98,7 +98,7 @@ pub use redshift::{
};
pub use s3::{S3SinkFixture, S3SinkOps, S3SinkRotationFixture};
pub use surrealdb::{
- SurrealDbOps, SurrealDbSinkBatchFixture, SurrealDbSinkFixture, SurrealDbSinkJsonFixture,
- SurrealDbSinkRawFixture,
+ SurrealDbOps, SurrealDbSinkBatchFixture, SurrealDbSinkDatabaseFixture, SurrealDbSinkFixture,
+ SurrealDbSinkJsonFixture, SurrealDbSinkNamespaceFixture, SurrealDbSinkRawFixture,
};
pub use wiremock::{WireMockDirectFixture, WireMockWrappedFixture};
diff --git a/core/integration/tests/connectors/fixtures/mongodb/sink.rs b/core/integration/tests/connectors/fixtures/mongodb/sink.rs
index 4052e0771..e33e1e039 100644
--- a/core/integration/tests/connectors/fixtures/mongodb/sink.rs
+++ b/core/integration/tests/connectors/fixtures/mongodb/sink.rs
@@ -240,8 +240,6 @@ impl TestFixture for MongoDbSinkJsonFixture {
fn connectors_runtime_envs(&self) -> HashMap<String, String> {
let mut envs = self.inner.connectors_runtime_envs();
- // Schema must be "json" for the runtime to route messages correctly.
- // Already set in base, but override STREAMS_0_SCHEMA explicitly.
envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string());
envs
}
diff --git a/core/integration/tests/connectors/fixtures/surrealdb/mod.rs b/core/integration/tests/connectors/fixtures/surrealdb/mod.rs
index 7e8e5fdad..9d3e4c43e 100644
--- a/core/integration/tests/connectors/fixtures/surrealdb/mod.rs
+++ b/core/integration/tests/connectors/fixtures/surrealdb/mod.rs
@@ -20,6 +20,6 @@ mod sink;
pub use container::SurrealDbOps;
pub use sink::{
- SurrealDbSinkBatchFixture, SurrealDbSinkFixture, SurrealDbSinkJsonFixture,
- SurrealDbSinkRawFixture,
+ SurrealDbSinkBatchFixture, SurrealDbSinkDatabaseFixture, SurrealDbSinkFixture,
+ SurrealDbSinkJsonFixture, SurrealDbSinkNamespaceFixture, SurrealDbSinkRawFixture,
};
diff --git a/core/integration/tests/connectors/fixtures/surrealdb/sink.rs b/core/integration/tests/connectors/fixtures/surrealdb/sink.rs
index 59f972933..a96241425 100644
--- a/core/integration/tests/connectors/fixtures/surrealdb/sink.rs
+++ b/core/integration/tests/connectors/fixtures/surrealdb/sink.rs
@@ -33,7 +33,11 @@ use std::time::Duration;
use tokio::time::sleep;
use tracing::info;
+const SCOPED_USERNAME: &str = "scoped_sink";
+const SCOPED_PASSWORD: &str = "scoped_password";
+
pub trait SurrealDbSinkProfile {
+ const AUTH_SCOPE: &'static str = "root";
const SCHEMA: &'static str;
const BATCH_SIZE: Option<usize>;
}
@@ -41,6 +45,8 @@ pub trait SurrealDbSinkProfile {
pub struct SurrealDbSinkJsonProfile;
pub struct SurrealDbSinkRawProfile;
pub struct SurrealDbSinkBatchProfile;
+pub struct SurrealDbSinkNamespaceProfile;
+pub struct SurrealDbSinkDatabaseProfile;
impl SurrealDbSinkProfile for SurrealDbSinkJsonProfile {
const SCHEMA: &'static str = "json";
@@ -57,6 +63,21 @@ impl SurrealDbSinkProfile for SurrealDbSinkBatchProfile {
const BATCH_SIZE: Option<usize> = Some(10);
}
+impl SurrealDbSinkProfile for SurrealDbSinkNamespaceProfile {
+ const AUTH_SCOPE: &'static str = "namespace";
+ const SCHEMA: &'static str = "json";
+ const BATCH_SIZE: Option<usize> = None;
+}
+
+impl SurrealDbSinkProfile for SurrealDbSinkDatabaseProfile {
+ const AUTH_SCOPE: &'static str = "database";
+ const SCHEMA: &'static str = "json";
+ const BATCH_SIZE: Option<usize> = None;
+}
+
+pub type SurrealDbSinkNamespaceFixture = SurrealDbSinkFixture<SurrealDbSinkNamespaceProfile>;
+pub type SurrealDbSinkDatabaseFixture = SurrealDbSinkFixture<SurrealDbSinkDatabaseProfile>;
+
pub type SurrealDbSinkJsonFixture = SurrealDbSinkFixture<SurrealDbSinkJsonProfile>;
pub type SurrealDbSinkRawFixture = SurrealDbSinkFixture<SurrealDbSinkRawProfile>;
pub type SurrealDbSinkBatchFixture = SurrealDbSinkFixture<SurrealDbSinkBatchProfile>;
@@ -169,6 +190,18 @@ where
{
async fn setup() -> Result<Self, TestBinaryError> {
let container = SurrealDbContainer::start().await?;
+ if P::AUTH_SCOPE != "root" {
+ let client = container.create_client().await?;
+ client
+ .query_result(&format!("DEFINE TABLE {DEFAULT_TABLE} SCHEMALESS;"))
+ .await?;
+ client
+ .query_result(&format!(
+ "DEFINE USER {SCOPED_USERNAME} ON {} PASSWORD '{SCOPED_PASSWORD}' ROLES EDITOR;",
+ P::AUTH_SCOPE
+ ))
+ .await?;
+ }
Ok(Self {
container,
profile: PhantomData,
@@ -187,11 +220,20 @@ where
);
envs.insert(ENV_SINK_DATABASE.to_string(), DEFAULT_DATABASE.to_string());
envs.insert(ENV_SINK_TABLE.to_string(), DEFAULT_TABLE.to_string());
- envs.insert(ENV_SINK_USERNAME.to_string(), ROOT_USERNAME.to_string());
- envs.insert(ENV_SINK_PASSWORD.to_string(), ROOT_PASSWORD.to_string());
- envs.insert(ENV_SINK_AUTH_SCOPE.to_string(), "root".to_string());
- envs.insert(ENV_SINK_AUTO_DEFINE_TABLE.to_string(), "true".to_string());
- envs.insert(ENV_SINK_DEFINE_INDEXES.to_string(), "true".to_string());
+ let root_auth = P::AUTH_SCOPE == "root";
+ let (username, password) = if root_auth {
+ (ROOT_USERNAME, ROOT_PASSWORD)
+ } else {
+ (SCOPED_USERNAME, SCOPED_PASSWORD)
+ };
+ envs.insert(ENV_SINK_USERNAME.to_string(), username.to_string());
+ envs.insert(ENV_SINK_PASSWORD.to_string(), password.to_string());
+ envs.insert(ENV_SINK_AUTH_SCOPE.to_string(), P::AUTH_SCOPE.to_string());
+ envs.insert(
+ ENV_SINK_AUTO_DEFINE_TABLE.to_string(),
+ root_auth.to_string(),
+ );
+ envs.insert(ENV_SINK_DEFINE_INDEXES.to_string(), root_auth.to_string());
envs.insert(ENV_SINK_PAYLOAD_FORMAT.to_string(), "auto".to_string());
envs.insert(
ENV_SINK_STREAMS_0_STREAM.to_string(),
diff --git a/core/integration/tests/connectors/influxdb/influxdb_sink.rs b/core/integration/tests/connectors/influxdb/influxdb_sink.rs
index f909e2770..90b364947 100644
--- a/core/integration/tests/connectors/influxdb/influxdb_sink.rs
+++ b/core/integration/tests/connectors/influxdb/influxdb_sink.rs
@@ -26,8 +26,7 @@ use integration::harness::seeds;
use integration::iggy_harness;
use serde_json::json;
-// seeds::connector_stream creates the topic with 1 partition (Iggy partition IDs are 1-based).
-// Use Partitioning::balanced() so the runtime picks partition 1 automatically.
+// The seed creates one partition; balanced partitioning selects it.
#[iggy_harness(
server(connectors_runtime(config_path = "tests/connectors/influxdb/sink.toml")),
@@ -230,8 +229,7 @@ async fn influxdb_sink_multiple_partitions(harness: &TestHarness, fixture: Influ
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- // Topic has only 1 partition — send 3 messages via balanced partitioning
- // (they all go to partition 1, which is correct for a 1-partition topic).
+ // All three messages use the single seeded partition.
for i in 1u32..=3 {
let payload = serde_json::to_vec(&json!({"msg_index": i})).expect("Failed to serialize");
let mut messages = vec![
diff --git a/core/integration/tests/connectors/mongodb/mongodb_sink.rs b/core/integration/tests/connectors/mongodb/mongodb_sink.rs
index b37975591..4aa4e500f 100644
--- a/core/integration/tests/connectors/mongodb/mongodb_sink.rs
+++ b/core/integration/tests/connectors/mongodb/mongodb_sink.rs
@@ -442,7 +442,7 @@ async fn duplicate_key_is_idempotent_replay_not_sink_error(
server(connectors_runtime(config_path = "tests/connectors/mongodb/sink.toml")),
seed = seeds::connector_stream
)]
-async fn ordered_duplicate_partial_insert_has_exact_accounting(
+async fn unordered_duplicate_partial_insert_has_exact_accounting(
harness: &TestHarness,
fixture: MongoDbSinkFixture,
) {
@@ -713,7 +713,7 @@ async fn schema_validation_mid_batch_surfaces_hard_error_and_partial_prefix(
);
assert!(
!id13_inserted,
- "Expected suffix message (_id=13) to be skipped after ordered schema validation failure"
+ "Expected suffix message (_id=13) to fail the collection validator"
);
let total_docs = collection
diff --git a/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs b/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
index 9205760c4..840810377 100644
--- a/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
+++ b/core/integration/tests/connectors/surrealdb/surrealdb_sink.rs
@@ -17,8 +17,8 @@
use super::{LARGE_BATCH_COUNT, POLL_ATTEMPTS, POLL_INTERVAL_MS, TEST_MESSAGE_COUNT};
use crate::connectors::fixtures::{
- SurrealDbOps, SurrealDbSinkBatchFixture, SurrealDbSinkFixture, SurrealDbSinkJsonFixture,
- SurrealDbSinkRawFixture,
+ SurrealDbOps, SurrealDbSinkBatchFixture, SurrealDbSinkDatabaseFixture, SurrealDbSinkFixture,
+ SurrealDbSinkJsonFixture, SurrealDbSinkNamespaceFixture, SurrealDbSinkRawFixture,
};
use bytes::Bytes;
use iggy::prelude::{IggyMessage, Partitioning};
@@ -57,6 +57,35 @@ fn push_hex_component(out: &mut String, bytes: &[u8]) {
seed = seeds::connector_stream
)]
async fn json_messages_sink_to_surrealdb(harness: &TestHarness, fixture: SurrealDbSinkJsonFixture) {
+ assert_json_messages(harness, fixture).await;
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/surrealdb/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_namespace_auth_should_insert_messages(
+ harness: &TestHarness,
+ fixture: SurrealDbSinkNamespaceFixture,
+) {
+ assert_json_messages(harness, fixture).await;
+}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/surrealdb/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_database_auth_should_insert_messages(
+ harness: &TestHarness,
+ fixture: SurrealDbSinkDatabaseFixture,
+) {
+ assert_json_messages(harness, fixture).await;
+}
+
+async fn assert_json_messages<P: Sync>(
+ harness: &integration::harness::TestHarness,
+ fixture: SurrealDbSinkFixture<P>,
+) {
let client = harness.root_client().await.unwrap();
let surreal_client = fixture
.create_client()