This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3c493f2f6 feat(connectors): deliver Iceberg sink messages at least
once (#4759)
3c493f2f6 is described below
commit 3c493f2f634e576ec03bde3493b531fa65f282c6
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Fri Nov 6 09:36:25 2026 +0100
feat(connectors): deliver Iceberg sink messages at least once (#4759)
---
.claude/skills/connector-sink/SKILL.md | 2 +-
core/connectors/runtime/README.md | 2 +-
.../example_config/connectors/iceberg_sink.toml | 4 +
core/connectors/sdk/README.md | 3 +-
core/connectors/sdk/src/lib.rs | 6 +-
core/connectors/sinks/iceberg_sink/Cargo.toml | 3 -
core/connectors/sinks/iceberg_sink/README.md | 90 +++-
.../sinks/iceberg_sink/src/router/commit.rs | 558 +++++++++++++++++++--
.../iceberg_sink/src/router/dynamic_router.rs | 303 ++++++++++-
.../sinks/iceberg_sink/src/router/mod.rs | 314 ++++++++----
.../sinks/iceberg_sink/src/router/redelivery.rs | 385 ++++++++++++++
.../sinks/iceberg_sink/src/router/static_router.rs | 254 +++++++++-
core/connectors/sinks/iceberg_sink/src/sink.rs | 9 +
.../src/harness/handle/connectors_runtime.rs | 20 +-
.../tests/connectors/fixtures/iceberg/container.rs | 139 ++++-
.../tests/connectors/fixtures/iceberg/mod.rs | 3 +-
core/integration/tests/connectors/fixtures/mod.rs | 3 +-
.../connectors/iceberg/iceberg_sink_delivery.rs | 229 +++++++++
core/integration/tests/connectors/iceberg/mod.rs | 1 +
19 files changed, 2125 insertions(+), 203 deletions(-)
diff --git a/.claude/skills/connector-sink/SKILL.md
b/.claude/skills/connector-sink/SKILL.md
index d93a2d10d..d9df101e4 100644
--- a/.claude/skills/connector-sink/SKILL.md
+++ b/.claude/skills/connector-sink/SKILL.md
@@ -96,7 +96,7 @@ Runtime retries on any `Err` - same batch can be redelivered.
In-tree guarantees
- **MongoDB**: composite `_id` from `stream:topic:partition:message_id`, via
`insert_many(docs).ordered(false)`. Per-document duplicate-key errors don't
abort the batch - effective idempotency under retry.
- **Elasticsearch** (`bulk_index_documents`): bulk action specifies only
`_index`, no `_id` -> ES auto-generates ids -> NOT idempotent. Extensions
should add `"_id"`.
- **HTTP**: endpoint-dependent. Document in the plugin README.
-- **Iceberg**: transactional commit. `Error::CatalogCommitError` documents
that the txn is consumed on `commit()`. Rebuild-then-retry can duplicate data -
callers must check the catalog first.
+- **Iceberg**: transactional commit per table. `router/commit.rs` tags each
snapshot with `iggy.commit-id`, and a commit retry first checks whether the
catalog already holds it. At least once, `router/redelivery.rs` records the
last message of each source partition in the snapshot summary. A redelivered
batch that carries the last message of any append on the current snapshot
history appends only the messages after the newest one.
When writing a new sink, prefer dedup-on-write using the message `id` field.
Mongo's upsert-style is the cleanest reference.
diff --git a/core/connectors/runtime/README.md
b/core/connectors/runtime/README.md
index d33aa17e7..18a3d63e6 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -139,7 +139,7 @@ At least once, the runtime behaves as follows:
The guarantee needs a plugin that supports it, and a sink configured for it
with a plugin that does not refuses to start. Of the bundled sinks, `http`
-supports it. It covers only messages the server still holds: use topic
+and `iceberg` support it. It covers only messages the server still holds: use
topic
`durability = "persisted"` and a retention longer than the longest expected
sink outage.
diff --git
a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml
b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml
index 143d7b893..4f411376b 100644
--- a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml
+++ b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml
@@ -51,6 +51,10 @@ max_retries = 5
retry_delay = "1s"
max_retry_delay = "30s"
+[delivery]
+# Store offsets only after the batch is committed to Iceberg.
+guarantee = "at_least_once"
+
[transforms.add_fields]
enabled = true
diff --git a/core/connectors/sdk/README.md b/core/connectors/sdk/README.md
index 53f9d7305..d073810c2 100644
--- a/core/connectors/sdk/README.md
+++ b/core/connectors/sdk/README.md
@@ -72,7 +72,8 @@ returns an error. The default refuses with
`Error::Unsupported`. Returning `Ok`
things:
1. A successful `consume()` never drops a message without failing the call. A
message that
- cannot be written fails the call instead of being skipped.
+ cannot be written fails the call instead of being skipped, unless the
sink's configuration
+ explicitly asks to skip such messages and the sink logs each one with its
offset.
2. A successful `Sink::flush()` proves that the destination acknowledged every
message of the
listed partitions that `consume()` accepted, up to and including each
listed offset.
diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs
index 21f19335c..b3decafc5 100644
--- a/core/connectors/sdk/src/lib.rs
+++ b/core/connectors/sdk/src/lib.rs
@@ -161,8 +161,10 @@ pub trait Sink: Send + Sync {
/// Called once after [`Sink::open`] when the sink is configured for
at-least-once delivery.
///
/// Returning `Ok` promises that, from now on, a successful
[`Sink::consume`] never drops a
- /// message without failing the call, and that a successful
[`Sink::flush`] proves that the
- /// destination acknowledged every message of the covered partitions
handed over so far.
+ /// message without failing the call, unless the sink's configuration
explicitly tells it to
+ /// skip such messages and it logs each one with its offset, and that a
successful
+ /// [`Sink::flush`] proves that the destination acknowledged every message
of the covered
+ /// partitions handed over so far.
/// Return [`Error::Unsupported`] when the sink, or the configuration it
was opened with,
/// cannot keep that promise. The default does.
fn enable_at_least_once(&mut self) -> Result<(), Error> {
diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml
b/core/connectors/sinks/iceberg_sink/Cargo.toml
index ab6141f1e..6a0f3ebaf 100644
--- a/core/connectors/sinks/iceberg_sink/Cargo.toml
+++ b/core/connectors/sinks/iceberg_sink/Cargo.toml
@@ -28,9 +28,6 @@ repository = "https://github.com/apache/iggy";
readme = "../../README.md"
publish = false
-[package.metadata.cargo-machete]
-ignored = ["dashmap"]
-
[lib]
crate-type = ["cdylib", "lib"]
diff --git a/core/connectors/sinks/iceberg_sink/README.md
b/core/connectors/sinks/iceberg_sink/README.md
index b0eade462..1b2bb3f20 100644
--- a/core/connectors/sinks/iceberg_sink/README.md
+++ b/core/connectors/sinks/iceberg_sink/README.md
@@ -9,6 +9,7 @@ The Iceberg Sink Connector allows you to consume messages from
Iggy topics and s
- **Single destination table**
- **Multiple-table fan-out static routing**
- **Multiple-table fan-out dynamic routing**
+- **At-least-once delivery**, with redelivered messages appended once (see
[Delivery Guarantee](#delivery-guarantee))
## Configuration example
@@ -50,7 +51,7 @@ max_retry_delay = "30s"
- **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.
-- **invalid_rows**: What to do with a row that does not match the table
schema. `fail`, the default, fails the whole batch. `skip` writes the rows that
match and logs each skipped row as a warning with its offset.
+- **invalid_rows**: What to do with a row that does not match the table
schema. `fail`, the default, fails the whole batch. `skip` writes the rows that
match and logs each skipped row as a warning with its offset. At least once, it
also decides what happens to a message that dynamic routing cannot route.
- **max_retries**: Total attempts per catalog commit, including the first.
Defaults to `5`. See [Commit Retries](#commit-retries).
- **retry_delay**: Delay before the first commit retry, doubled for each
further retry. Defaults to `1s`.
- **max_retry_delay**: Upper bound for one delay between commit attempts.
Defaults to `30s`.
@@ -60,24 +61,31 @@ Static routing requires a non-empty `tables` list; dynamic
routing requires a no
## 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.
+With `dynamic_routing = false`, every batch is copied to every table in
`tables`. Startup fails
+if none of them loads. A table that does not load at startup is loaded again
by each batch. At
+most once, a batch skips a table that still does not load. At least once, it
fails instead:
+transiently while the catalog cannot be reached, so that the runtime delivers
it again, and for
+good when the table does not exist. A name without a namespace is skipped at
startup, and at
+least once the sink refuses to start with one. The writer uses the schema and
default partition
+spec captured when the table loaded, 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 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:
+A message cannot be routed in the following cases:
-- 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, and
every skipped
-message is logged as a warning with its offset. Each batch loads its
destination tables again. The route field remains in the row, but the writer
ignores it if the
+- The table name has no namespace or contains an empty name component.
+- The table named in the message field, or its namespace, does not exist in
the catalog.
+
+At most once, such a message is skipped, and so is a message whose table does
not load for any
+other reason, with a warning naming its offset. At least once, a message that
cannot be routed
+stops the call at that message, unless `invalid_rows = "skip"` skips it the
same way. Any other
+failure to load a table, such as an unreachable catalog or denied access,
fails the call without
+skipping anything (see [Delivery Guarantee](#delivery-guarantee)). 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
@@ -122,9 +130,10 @@ Each destination table has its own append transaction.
Fan-out is not atomic acr
a write or commit error stops work on the remaining tables, while earlier
commits remain.
The Iceberg library refreshes metadata before committing and retries commit
conflicts
according to the table's retry properties, and the sink retries other
transient commit errors.
-Once those retries run out, 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.
+At most once, once those retries run out, 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. At least once, the runtime
delivers the batch
+again, and each table appends only the messages it does not hold yet.
## Commit Retries
@@ -140,8 +149,53 @@ was lost is not appended twice. A commit request that the
catalog applies only a
checked for it, for example behind a gateway that timed out while the catalog
kept working, can
still be appended twice.
-When the attempts run out, the batch fails and is lost, because the runtime
stored its offset when
-it read the batch. Raise `max_retries` or `max_retry_delay` to ride out a
longer catalog outage.
+When the attempts run out, the batch fails. At most once, it is lost, because
the runtime stored
+its offset when it read the batch: raise `max_retries` or `max_retry_delay` to
ride out a longer
+catalog outage. At least once, the runtime delivers the batch again after its
own backoff.
+
+## Delivery Guarantee
+
+The sink delivers at most once unless the connector's `[delivery]` table asks
for at least once:
+
+```toml
+[delivery]
+guarantee = "at_least_once"
+```
+
+The [runtime README](../../runtime/README.md#sink-delivery-guarantees)
describes the table, the
+retries and how a stopped delivery resumes. At least once:
+
+- **What a successful call proves.** The sink writes and commits each batch
inside `consume`, so a
+ call returns only after the catalog committed the batch to every destination
table. The runtime
+ stores the offset only after that, so a crash, a failed write, or an outage
of the catalog or the
+ object store delivers the batch again instead of losing it.
+- **Failures.** A write or commit that fails after the sink's own retries, and
a catalog that cannot
+ be reached, fail the call transiently, and the runtime delivers the batch
again after its
+ backoff. A commit that the catalog refuses for good, such as one denied by
its permissions, and a
+ table that does not exist stop the delivery of the topic.
+- **Rejected messages.** A row that does not match the table schema, a payload
that holds no JSON
+ document, and a message that dynamic routing cannot route stop the delivery
of the topic at
+ that message. The messages before it are committed first, and the error
names its offset, so
+ storing that offset, as the runtime README describes, gives up that message
alone.
+ `invalid_rows = "skip"` skips such a message instead, with a warning naming
its offset. Skipping
+ is an explicit choice that drops the message, and the runtime stores offsets
past it.
+- **Redelivered messages are appended once.** Each append records in its
snapshot summary the
+ source of its rows (`iggy.stream`, `iggy.topic`, `iggy.partition-id`) and
its last message
+ (`iggy.last-offset`, `iggy.last-message-id`, and `iggy.last-timestamp`, the
time the server
+ stored it). When a batch arrives again, and it carries the last message of
any append on the
+ table's current snapshot history, the sink appends only the messages after
the newest append.
+ A crash between the commit and the offset store, a backlog of several
unflushed batches, or a
+ batch that committed to one table and failed on another, therefore does not
append rows twice.
+ Each table keeps its own record, so a batch copied to several tables is
appended once to each.
+- **Limits.** The record is the newest message a table holds, not a list of
its rows. A message at
+ or below it is skipped even when the table lacks its row, for example a row
deleted later, or
+ one that `invalid_rows = "skip"` left out. To append such messages again,
replay them with
+ `guarantee = "at_most_once"`, which does not skip anything. A batch that
carries no appended last
+ message, for example after `batch_length` changed, cannot be proven and is
appended in full.
+ After a rollback of the table, restart the sink, since it keeps the record
of its own commits
+ in memory. Keep one sink writing each table and partition, and keep the
stream and topic
+ identifiers in its configuration unchanged, since the record names them. At
most once, appends
+ record nothing.
## Commit Rate and Table Maintenance
@@ -164,6 +218,10 @@ At 2,000 messages per second, `batch_length = 100` creates
about 20 snapshots pe
values above about 2. A poll also returns once its reply reaches 1 MiB, and
`max_wait` bounds how
long a message waits for its batch.
+At least once, each snapshot summary also carries the six `iggy.*` properties
described in
+[Delivery Guarantee](#delivery-guarantee), which add a few hundred bytes per
snapshot to the table
+metadata until the snapshot expires.
+
The sink does not expire snapshots or rewrite manifests. Run that maintenance
from an engine that
supports it, for example Spark's `expire_snapshots` and `rewrite_manifests`
procedures, on a
schedule that keeps the time travel you need. Some managed catalogs run it
automatically.
diff --git a/core/connectors/sinks/iceberg_sink/src/router/commit.rs
b/core/connectors/sinks/iceberg_sink/src/router/commit.rs
index 801fa76d0..768ecc31e 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/commit.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/commit.rs
@@ -31,25 +31,25 @@ use uuid::Uuid;
/// whether the catalog applied an earlier attempt whose response was lost.
pub(crate) const COMMIT_ID_PROPERTY: &str = "iggy.commit-id";
-/// Appends `data_files` to `table` in one snapshot and retries a failed commit
-/// with `retry`. Every retry first looks for the snapshot of an earlier
attempt.
+/// Appends `data_files` to `table` in one snapshot, with `properties` in its
summary, and
+/// retries a failed commit with `retry`. Every retry first looks for the
snapshot of an earlier
+/// attempt.
pub(crate) async fn commit_data_files(
table: &Table,
catalog: &dyn Catalog,
data_files: Vec<DataFile>,
+ mut properties: HashMap<String, String>,
retry: RetryPolicy,
) -> Result<(), Error> {
let commit_id = Uuid::new_v4().to_string();
+ properties.insert(COMMIT_ID_PROPERTY.to_owned(), commit_id.clone());
let transaction = Transaction::new(table);
// The check reads every manifest of the table at once on each commit, and
the data files
// carry fresh UUID names it could never match.
let action = transaction
.fast_append()
.with_check_duplicate(false)
- .set_snapshot_properties(HashMap::from([(
- COMMIT_ID_PROPERTY.to_owned(),
- commit_id.clone(),
- )]))
+ .set_snapshot_properties(properties)
.add_data_files(data_files);
let transaction = action.apply(transaction).map_err(|err| {
let chain = format_error_chain(&err);
@@ -82,17 +82,50 @@ pub(crate) async fn commit_data_files(
failure.attempts,
chain
);
- Error::CatalogCommitError(chain)
+ // A conflict outlasted the table's own retries, so a delivery of the
batch again rebuilds
+ // the transaction on fresh metadata. Any other refusal, such as a
denied write, fails the
+ // same way every time.
+ if is_transient(&failure.error) || failure.error.kind() ==
ErrorKind::CatalogCommitConflicts
+ {
+ Error::CatalogCommitError(chain)
+ } else {
+ Error::PermanentHttpError(format!(
+ "The Iceberg catalog refused the commit to table {identifier}:
{chain}"
+ ))
+ }
})
}
-/// A commit that could not reach the catalog, or that the catalog failed or
throttled, can pass
-/// on another attempt. A conflict is not retried here: `Transaction::commit`
has already retried
-/// it as far as the table's `commit.retry.*` properties allow.
-fn is_transient(error: &iceberg::Error) -> bool {
+/// A catalog call that could not reach the catalog, or that the catalog
failed or throttled, can
+/// pass on another attempt. A commit conflict is not retried here:
`Transaction::commit` has
+/// already retried it as far as the table's `commit.retry.*` properties allow.
+pub(crate) fn is_transient(error: &iceberg::Error) -> bool {
error.kind() == ErrorKind::Unexpected &&
response_status(error).is_none_or(is_transient_status)
}
+/// The error for a table that does not load from the catalog. An unreachable,
failing or
+/// throttling catalog is transient. Any other refusal, such as a missing
table or denied access,
+/// fails the same way until the configuration or the catalog changes.
+pub(crate) fn table_load_error(table: &str, error: &iceberg::Error) -> Error {
+ let reason = format!(
+ "Failed to load Iceberg table {table} from the catalog: {}",
+ format_error_chain(error)
+ );
+ if is_transient(error) {
+ Error::Connection(reason)
+ } else {
+ Error::InvalidConfigValue(reason)
+ }
+}
+
+/// Whether a table load failed because the table or its namespace does not
exist.
+pub(crate) fn is_missing(error: &iceberg::Error) -> bool {
+ matches!(
+ error.kind(),
+ ErrorKind::TableNotFound | ErrorKind::NamespaceNotFound
+ )
+}
+
/// `iceberg-catalog-rest` keeps the status of an answer it did not expect
only in the error's
/// private context, which `Display` renders first: `context: { status: 403
Forbidden, ...` for a
/// catalog call and `context: { code: 401 Unauthorized, ...` for the token
endpoint.
@@ -136,18 +169,24 @@ fn holds_commit(table: &Table, commit_id: &str) -> bool {
}
#[cfg(test)]
-mod tests {
+pub(crate) mod tests {
use super::*;
+ use crate::InvalidRows;
+ use crate::router::redelivery::{HeldMessages, LastMessage,
snapshot_properties};
+ use crate::router::{BatchOrigin, Written, write_data};
use async_trait::async_trait;
use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use iceberg::spec::{
DataContentType, DataFileBuilder, DataFileFormat, NestedField,
PrimitiveType,
Schema as IcebergSchema, Type,
};
+ use iceberg::spec::{MAIN_BRANCH, SnapshotReference, SnapshotRetention,
TableMetadataBuilder};
use iceberg::{
CatalogBuilder, MemoryCatalog, Namespace, NamespaceIdent, TableCommit,
TableCreation,
};
use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestCatalog,
RestCatalogBuilder};
+ use iggy_connector_sdk::Schema;
+ use iggy_connector_sdk::{ConsumedMessage, Payload};
use std::collections::VecDeque;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU32, Ordering};
@@ -387,9 +426,15 @@ mod tests {
FlakyCatalog::with_table(&[CommitFailure::Unreachable,
CommitFailure::Unreachable])
.await;
- commit_data_files(&table, &catalog, vec![data_file(&table)],
FAST_RETRY)
- .await
- .expect("The third attempt should commit");
+ commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect("The third attempt should commit");
assert_eq!(catalog.update_calls(), 3);
assert_eq!(catalog.snapshot_count(&table).await, 1);
@@ -399,9 +444,15 @@ mod tests {
async fn given_a_lost_commit_response_should_not_append_the_files_twice() {
let (catalog, table) =
FlakyCatalog::with_table(&[CommitFailure::ResponseLost]).await;
- commit_data_files(&table, &catalog, vec![data_file(&table)],
FAST_RETRY)
- .await
- .expect("The retry should find the applied commit");
+ commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect("The retry should find the applied commit");
assert_eq!(catalog.update_calls(), 1);
assert_eq!(catalog.snapshot_count(&table).await, 1);
@@ -411,11 +462,20 @@ mod tests {
async fn given_a_refused_commit_should_fail_without_retrying() {
let (catalog, table) =
FlakyCatalog::with_table(&[CommitFailure::Refused]).await;
- let error = commit_data_files(&table, &catalog,
vec![data_file(&table)], FAST_RETRY)
- .await
- .expect_err("A refused commit should fail");
-
- assert!(matches!(error, Error::CatalogCommitError(_)), "{error}");
+ let error = commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect_err("A refused commit should fail");
+
+ assert!(
+ error.is_permanent(),
+ "a refused commit should stop the delivery: {error}"
+ );
assert_eq!(catalog.update_calls(), 1);
assert_eq!(catalog.snapshot_count(&table).await, 0);
}
@@ -425,11 +485,20 @@ mod tests {
let (catalog, table) =
FlakyCatalog::with_table(&[CommitFailure::Rejected(StatusCode::FORBIDDEN)]).await;
- let error = commit_data_files(&table, &catalog,
vec![data_file(&table)], FAST_RETRY)
- .await
- .expect_err("A forbidden commit should fail");
-
- assert!(matches!(error, Error::CatalogCommitError(_)), "{error}");
+ let error = commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect_err("A forbidden commit should fail");
+
+ assert!(
+ error.is_permanent(),
+ "a refused commit should stop the delivery: {error}"
+ );
assert_eq!(catalog.update_calls(), 1);
assert_eq!(catalog.snapshot_count(&table).await, 0);
}
@@ -440,9 +509,15 @@ mod tests {
FlakyCatalog::with_table(&[CommitFailure::Rejected(StatusCode::TOO_MANY_REQUESTS)])
.await;
- commit_data_files(&table, &catalog, vec![data_file(&table)],
FAST_RETRY)
- .await
- .expect("The retry should commit");
+ commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect("The retry should commit");
assert_eq!(catalog.update_calls(), 2);
assert_eq!(catalog.snapshot_count(&table).await, 1);
@@ -458,9 +533,15 @@ mod tests {
let (catalog, table) =
FlakyCatalog::with_table_properties(&[CommitFailure::Conflict; 6],
table_retries).await;
- let error = commit_data_files(&table, &catalog,
vec![data_file(&table)], FAST_RETRY)
- .await
- .expect_err("The commit should fail once the table retries run
out");
+ let error = commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect_err("The commit should fail once the table retries run out");
assert!(matches!(error, Error::CatalogCommitError(_)), "{error}");
assert_eq!(catalog.update_calls(), 2);
@@ -491,9 +572,15 @@ mod tests {
async fn
given_an_unreachable_catalog_when_the_attempts_run_out_should_fail() {
let (catalog, table) =
FlakyCatalog::with_table(&[CommitFailure::Unreachable; 3]).await;
- let error = commit_data_files(&table, &catalog,
vec![data_file(&table)], FAST_RETRY)
- .await
- .expect_err("The commit should fail once the attempts run out");
+ let error = commit_data_files(
+ &table,
+ &catalog,
+ vec![data_file(&table)],
+ HashMap::new(),
+ FAST_RETRY,
+ )
+ .await
+ .expect_err("The commit should fail once the attempts run out");
assert!(matches!(error, Error::CatalogCommitError(_)), "{error}");
assert_eq!(catalog.update_calls(), FAST_RETRY.max_attempts);
@@ -501,7 +588,8 @@ mod tests {
}
/// A REST catalog whose server answers every request with `status` and no
body.
- async fn rest_catalog_answering(status: StatusCode) -> RestCatalog {
+ /// A REST catalog that answers every call with `status`.
+ pub(crate) async fn rest_catalog_answering(status: StatusCode) ->
RestCatalog {
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("Failed to bind the catalog server");
@@ -528,4 +616,398 @@ mod tests {
.await
.expect("Failed to create the REST catalog")
}
+
+ const ORIGIN: BatchOrigin<'static> = BatchOrigin {
+ connector_id: 1,
+ stream: "stream",
+ topic: "topic",
+ partition_id: 2,
+ };
+
+ fn message(offset: u64) -> ConsumedMessage {
+ ConsumedMessage {
+ id: u128::from(offset) + 100,
+ offset,
+ checksum: 0,
+ timestamp: offset + 1_000,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(Vec::new()),
+ }
+ }
+
+ fn messages(offsets: std::ops::RangeInclusive<u64>) ->
Vec<ConsumedMessage> {
+ offsets.map(message).collect()
+ }
+
+ fn json_messages(offsets: std::ops::RangeInclusive<u64>) ->
Vec<ConsumedMessage> {
+ offsets
+ .map(|offset| ConsumedMessage {
+ payload: Payload::Json(simd_json::json!({ "id": offset })),
+ ..message(offset)
+ })
+ .collect()
+ }
+
+ fn offsets(messages: &[ConsumedMessage]) -> Vec<u64> {
+ messages.iter().map(|message| message.offset).collect()
+ }
+
+ async fn commit_up_to(
+ catalog: &FlakyCatalog,
+ table: &Table,
+ origin: &BatchOrigin<'_>,
+ offset: u64,
+ ) {
+ let properties = snapshot_properties(origin,
LastMessage::from(&message(offset)));
+ commit_data_files(
+ table,
+ catalog,
+ vec![data_file(table)],
+ properties,
+ FAST_RETRY,
+ )
+ .await
+ .expect("The append should commit");
+ }
+
+ async fn current_table(catalog: &FlakyCatalog, table: &Table) -> Table {
+ catalog
+ .inner
+ .load_table(table.identifier())
+ .await
+ .expect("Failed to load table")
+ }
+
+ #[allow(clippy::too_many_arguments)]
+ async fn write(
+ catalog: &FlakyCatalog,
+ table: &Table,
+ batch: &[ConsumedMessage],
+ invalid_rows: InvalidRows,
+ retry: RetryPolicy,
+ held: Option<&HeldMessages>,
+ ) -> Result<Written, Error> {
+ write_data(
+ &ORIGIN,
+ batch,
+ table,
+ None,
+ catalog,
+ Schema::Json,
+ invalid_rows,
+ retry,
+ held,
+ )
+ .await
+ }
+
+ #[tokio::test]
+ async fn
given_a_table_holding_part_of_a_batch_should_append_only_the_messages_after_it()
{
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+ commit_up_to(&catalog, &table, &ORIGIN, 4).await;
+ let other_partition = BatchOrigin {
+ partition_id: 3,
+ ..ORIGIN
+ };
+ commit_up_to(&catalog, &table, &other_partition, 9).await;
+ let batch = messages(3..=5);
+
+ let unheld = HeldMessages::default()
+ .unheld(&ORIGIN, &table, None, &catalog, &batch)
+ .await
+ .expect("The table should load");
+
+ assert_eq!(offsets(unheld), [5]);
+ }
+
+ #[tokio::test]
+ async fn
given_a_backlog_batch_carrying_an_older_append_should_skip_up_to_the_newest() {
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+ for offset in [1, 3, 5] {
+ commit_up_to(&catalog, &table, &ORIGIN, offset).await;
+ }
+ let first_batch = messages(0..=1);
+ let longer_batch = messages(0..=6);
+
+ let held = HeldMessages::default();
+ let first = held
+ .unheld(&ORIGIN, &table, None, &catalog, &first_batch)
+ .await
+ .expect("The table should load");
+ let longer = held
+ .unheld(&ORIGIN, &table, None, &catalog, &longer_batch)
+ .await
+ .expect("The table should load");
+
+ assert!(first.is_empty(), "{:?}", offsets(first));
+ assert_eq!(offsets(longer), [6]);
+ }
+
+ #[tokio::test]
+ async fn
given_a_rolled_back_table_should_not_count_the_appends_left_behind() {
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+ commit_up_to(&catalog, &table, &ORIGIN, 4).await;
+ let kept = current_table(&catalog, &table)
+ .await
+ .metadata()
+ .current_snapshot_id()
+ .expect("The append should be the current snapshot");
+ commit_up_to(&catalog, &table, &ORIGIN, 9).await;
+ let current = current_table(&catalog, &table).await;
+ let rolled_back =
TableMetadataBuilder::new_from_metadata(current.metadata().clone(), None)
+ .set_ref(
+ MAIN_BRANCH,
+ SnapshotReference::new(kept, SnapshotRetention::branch(None,
None, None)),
+ )
+ .expect("The branch should move back")
+ .build()
+ .expect("The metadata should build")
+ .metadata;
+ let rolled_back = Table::builder()
+ .metadata(rolled_back)
+ .identifier(current.identifier().clone())
+ .file_io(current.file_io().clone())
+ .runtime(iceberg::Runtime::current())
+ .build()
+ .expect("The table should build");
+ let batch = messages(3..=10);
+
+ let unheld = HeldMessages::default()
+ .unheld(&ORIGIN, &rolled_back, Some(&rolled_back), &catalog,
&batch)
+ .await
+ .expect("The current table needs no catalog");
+
+ assert_eq!(offsets(unheld), [5, 6, 7, 8, 9, 10]);
+ }
+
+ #[tokio::test]
+ async fn
given_a_commit_this_sink_made_should_skip_it_without_loading_the_table() {
+ let (_, table) = FlakyCatalog::with_table(&[]).await;
+ let unreachable =
rest_catalog_answering(StatusCode::SERVICE_UNAVAILABLE).await;
+ let held = HeldMessages::default();
+ held.committed(&ORIGIN, &table, LastMessage::from(&message(4)));
+ let batch = messages(3..=5);
+
+ let unheld = held
+ .unheld(&ORIGIN, &table, None, &unreachable, &batch)
+ .await
+ .expect("The known commit should need no catalog");
+
+ assert_eq!(offsets(unheld), [5]);
+ }
+
+ #[tokio::test]
+ async fn given_a_failed_commit_should_read_the_table_again() {
+ let (_, table) = FlakyCatalog::with_table(&[]).await;
+ let unreachable =
rest_catalog_answering(StatusCode::SERVICE_UNAVAILABLE).await;
+ let held = HeldMessages::default();
+ held.committed(&ORIGIN, &table, LastMessage::from(&message(4)));
+ held.forget(&ORIGIN, &table);
+
+ let error = held
+ .unheld(&ORIGIN, &table, None, &unreachable, &messages(3..=5))
+ .await
+ .expect_err("A forgotten commit should be read from the catalog");
+
+ assert!(!error.is_permanent(), "{error}");
+ }
+
+ #[tokio::test]
+ async fn
given_a_catalog_that_cannot_load_the_table_should_fail_by_the_cause() {
+ let (_, table) = FlakyCatalog::with_table(&[]).await;
+ for (status, permanent) in [
+ (StatusCode::SERVICE_UNAVAILABLE, false),
+ (StatusCode::TOO_MANY_REQUESTS, false),
+ (StatusCode::FORBIDDEN, true),
+ (StatusCode::NOT_FOUND, true),
+ ] {
+ let catalog = rest_catalog_answering(status).await;
+
+ let error = HeldMessages::default()
+ .unheld(&ORIGIN, &table, None, &catalog, &messages(3..=5))
+ .await
+ .expect_err("The table should not load");
+
+ assert_eq!(error.is_permanent(), permanent, "{status}: {error}");
+ }
+ }
+
+ #[tokio::test]
+ async fn
given_a_batch_below_the_newest_append_when_recording_should_keep_the_newest() {
+ let (_, table) = FlakyCatalog::with_table(&[]).await;
+ let held = HeldMessages::default();
+ held.committed(&ORIGIN, &table, LastMessage::from(&message(9)));
+
+ let lower = held.last_to_record(&ORIGIN, &table,
LastMessage::from(&message(3)));
+ let higher = held.last_to_record(&ORIGIN, &table,
LastMessage::from(&message(12)));
+
+ assert_eq!(lower, LastMessage::from(&message(9)));
+ assert_eq!(higher, LastMessage::from(&message(12)));
+ }
+
+ #[tokio::test]
+ async fn
given_a_commit_that_landed_but_failed_when_delivered_again_should_not_append_it_twice()
+ {
+ let (catalog, table) =
FlakyCatalog::with_table(&[CommitFailure::ResponseLost]).await;
+ let single_attempt = RetryPolicy {
+ max_attempts: 1,
+ ..FAST_RETRY
+ };
+ let held = HeldMessages::default();
+ let batch = json_messages(0..=2);
+
+ write(
+ &catalog,
+ &table,
+ &batch,
+ InvalidRows::Fail,
+ single_attempt,
+ Some(&held),
+ )
+ .await
+ .expect_err("The lost response should fail the commit");
+ let again = write(
+ &catalog,
+ &table,
+ &batch,
+ InvalidRows::Fail,
+ single_attempt,
+ Some(&held),
+ )
+ .await
+ .expect("The batch delivered again should succeed");
+
+ assert_eq!(again.committed_rows, 0, "the table already holds the
batch");
+ assert_eq!(catalog.snapshot_count(&table).await, 1);
+ }
+
+ #[tokio::test]
+ async fn
given_a_row_that_does_not_match_when_failing_rows_should_commit_the_rows_before_it()
{
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+ let mut batch = json_messages(0..=3);
+ batch[2].payload = Payload::Json(simd_json::json!({ "id": "two" }));
+
+ let written = write(
+ &catalog,
+ &table,
+ &batch,
+ InvalidRows::Fail,
+ FAST_RETRY,
+ Some(&HeldMessages::default()),
+ )
+ .await
+ .expect("The rows before the rejected one should commit");
+
+ assert_eq!(written.committed_rows, 2);
+ let rejection = written.rejection.expect("The row should be rejected");
+ assert_eq!(rejection.offset, 2);
+ assert!(rejection.error.is_permanent(), "{}", rejection.error);
+ assert!(
+ rejection.error.to_string().contains("offset: 2"),
+ "{}",
+ rejection.error
+ );
+ assert_eq!(catalog.snapshot_count(&table).await, 1);
+ }
+
+ #[tokio::test]
+ async fn
given_a_payload_without_json_should_reject_it_or_skip_it_as_configured() {
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+ let mut batch = json_messages(0..=3);
+ batch[1].payload = Payload::Proto("id: 1".to_owned());
+
+ let failed = write(
+ &catalog,
+ &table,
+ &batch,
+ InvalidRows::Fail,
+ FAST_RETRY,
+ None,
+ )
+ .await
+ .expect("The rows before the rejected one should commit");
+ let skipped = write(
+ &catalog,
+ &table,
+ &batch,
+ InvalidRows::Skip,
+ FAST_RETRY,
+ None,
+ )
+ .await
+ .expect("Skipping should write the other rows");
+
+ assert_eq!(failed.committed_rows, 1);
+ assert_eq!(failed.rejection.map(|rejection| rejection.offset),
Some(1));
+ assert_eq!(skipped.committed_rows, 3);
+ assert!(skipped.rejection.is_none());
+ }
+
+ #[tokio::test]
+ async fn
given_at_most_once_should_record_no_delivered_messages_in_the_snapshot() {
+ let (catalog, table) = FlakyCatalog::with_table(&[]).await;
+
+ write(
+ &catalog,
+ &table,
+ &json_messages(0..=2),
+ InvalidRows::Fail,
+ FAST_RETRY,
+ None,
+ )
+ .await
+ .expect("The batch should commit");
+
+ let current = current_table(&catalog, &table).await;
+ let snapshot = current
+ .metadata()
+ .current_snapshot()
+ .expect("The batch should be appended");
+ let properties = &snapshot.summary().additional_properties;
+ assert!(properties.contains_key(COMMIT_ID_PROPERTY));
+ assert!(
+ !properties.keys().any(|key| key.starts_with("iggy.last-")),
+ "{properties:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn given_an_empty_call_should_succeed_without_the_catalog() {
+ let (_, table) = FlakyCatalog::with_table(&[]).await;
+ let unreachable =
rest_catalog_answering(StatusCode::SERVICE_UNAVAILABLE).await;
+
+ let written = write_data(
+ &ORIGIN,
+ &[],
+ &table,
+ None,
+ &unreachable,
+ Schema::Json,
+ InvalidRows::Fail,
+ FAST_RETRY,
+ Some(&HeldMessages::default()),
+ )
+ .await
+ .expect("An empty call has nothing to write");
+
+ assert_eq!(written.committed_rows, 0);
+ }
+
+ #[test]
+ fn given_load_errors_should_tell_missing_tables_and_outages_apart() {
+ let missing = iceberg::Error::new(ErrorKind::TableNotFound, "Table
does not exist");
+ let unreachable = iceberg::Error::new(ErrorKind::Unexpected,
"Connection refused");
+ let forbidden = iceberg::Error::new(
+ ErrorKind::Unexpected,
+ "Received response with unexpected status code",
+ )
+ .with_context("status", StatusCode::FORBIDDEN.to_string());
+
+ assert!(is_missing(&missing));
+ assert!(!is_missing(&forbidden));
+ assert!(table_load_error("test.events", &missing).is_permanent());
+ assert!(!table_load_error("test.events", &unreachable).is_permanent());
+ assert!(table_load_error("test.events", &forbidden).is_permanent());
+ }
}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
index 64bc1d113..57f035cb5 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs
@@ -16,12 +16,17 @@
// under the License.
use crate::InvalidRows;
-use crate::router::{BatchOrigin, Router, is_valid_namespaced_table,
table_exists, write_data};
+use crate::router::commit::{is_missing, table_load_error};
+use crate::router::redelivery::HeldMessages;
+use crate::router::{
+ BatchOrigin, Rejection, Router, before, first_rejection,
format_error_chain,
+ is_valid_namespaced_table, load_table, write_data,
+};
use async_trait::async_trait;
use iceberg::Catalog;
use iceberg::table::Table;
use iggy_connector_sdk::retry::RetryPolicy;
-use iggy_connector_sdk::{ConsumedMessage, Payload, Schema};
+use iggy_connector_sdk::{ConsumedMessage, Error, Payload, Schema};
use simd_json::base::ValueAsObject;
use std::collections::HashMap;
use std::collections::hash_map::Entry;
@@ -33,6 +38,7 @@ pub struct DynamicRouter {
route_field: String,
invalid_rows: InvalidRows,
commit_retry: RetryPolicy,
+ held: Option<HeldMessages>,
}
impl DynamicRouter {
@@ -47,12 +53,37 @@ impl DynamicRouter {
route_field,
invalid_rows,
commit_retry,
+ held: None,
}
}
fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String>
{
route_value(&message.payload, &self.route_field)
}
+
+ /// Drops a message that names no table it can be written to. At least
once, it rejects the
+ /// message instead, unless `invalid_rows` skips rows, so that the call
stops at it.
+ fn unroutable(
+ &self,
+ origin: &BatchOrigin<'_>,
+ message: &ConsumedMessage,
+ reason: &str,
+ ) -> Option<Rejection> {
+ if self.held.is_some() && self.invalid_rows == InvalidRows::Fail {
+ return Some(Rejection {
+ offset: message.offset,
+ error: Error::InvalidRecordValue(format!(
+ "The message at offset {} ({origin}) cannot be routed:
{reason}",
+ message.offset
+ )),
+ });
+ }
+ warn!(
+ "Dropping the message at offset {} ({origin}): {reason}",
+ message.offset
+ );
+ None
+ }
}
/// The routing value a payload carries under `route_field`, read from the JSON
@@ -76,17 +107,31 @@ impl Router for DynamicRouter {
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error> {
let mut tables: HashMap<String, (Table, Vec<ConsumedMessage>)> =
HashMap::new();
+ // Route values that name no table, with the reason, so that a batch
loads each once.
+ let mut unroutable: HashMap<String, String> = HashMap::new();
+ let mut rejection = None;
for mut message in messages {
// Normalised once here so the routing read and the later write
both
// work off the document, rather than parsing the same proto text
twice.
message.payload = message.payload.into_json_document();
let Some(route_field_val) = self.extract_route_field(&message)
else {
- warn!(
- "Dropping the message at offset {}: it is not a JSON
object with the route field: {}",
- message.offset, self.route_field
+ let reason = format!(
+ "it is not a JSON object with the route field: {}",
+ self.route_field
);
+ rejection = self.unroutable(origin, &message, &reason);
+ if rejection.is_some() {
+ break;
+ }
continue;
};
+ if let Some(reason) = unroutable.get(&route_field_val) {
+ rejection = self.unroutable(origin, &message, reason);
+ if rejection.is_some() {
+ break;
+ }
+ continue;
+ }
let entry = match tables.entry(route_field_val) {
Entry::Occupied(mut entry) => {
@@ -97,49 +142,192 @@ impl Router for DynamicRouter {
};
let route_field_val = entry.key();
- if !is_valid_namespaced_table(route_field_val) {
- warn!(
- "Dropping the message at offset {}: its route field value:
{route_field_val} names no table. It needs at least one namespace before the
table, separated by '.'",
- message.offset
- );
- continue;
- }
-
- let Some(table) = table_exists(route_field_val,
self.catalog.as_ref()).await else {
- warn!(
- "Dropping the message at offset {}: table:
{route_field_val} does not load from the catalog",
- message.offset
- );
- continue;
+ let reason = if is_valid_namespaced_table(route_field_val) {
+ match load_table(route_field_val, self.catalog.as_ref()).await
{
+ Ok(table) => {
+ entry.insert((table, vec![message]));
+ continue;
+ }
+ Err(error) if is_missing(&error) => format!(
+ "table: {route_field_val} does not exist in the
catalog: {}",
+ format_error_chain(&error)
+ ),
+ // An unreachable catalog or a refused load says nothing
about the message,
+ // so at least once the batch fails without skipping it.
+ Err(error) if self.held.is_some() => {
+ return Err(table_load_error(route_field_val, &error));
+ }
+ Err(error) => format!(
+ "table: {route_field_val} does not load from the
catalog: {}",
+ format_error_chain(&error)
+ ),
+ }
+ } else {
+ format!(
+ "its route field value: {route_field_val} names no table.
It needs at least one namespace before the table, separated by '.'"
+ )
};
- entry.insert((table, vec![message]));
+ rejection = self.unroutable(origin, &message, &reason);
+ if rejection.is_some() {
+ break;
+ }
+ unroutable.insert(entry.into_key(), reason);
}
for (table_name, (table, batch_messages)) in tables {
- let committed_rows = write_data(
+ // Once a message is rejected, the tables still need every message
before it.
+ let messages = before(&batch_messages, rejection.as_ref());
+ let written = write_data(
origin,
- &batch_messages,
+ messages,
&table,
+ Some(&table),
self.catalog.as_ref(),
messages_schema,
self.invalid_rows,
self.commit_retry,
+ self.held.as_ref(),
)
.await?;
info!(
- "Dynamically routed {} input rows to iceberg table
{table_name}, committed {committed_rows}",
- batch_messages.len()
+ "Dynamically routed {} input rows to iceberg table
{table_name}, committed {}",
+ messages.len(),
+ written.committed_rows
);
+ rejection = first_rejection(rejection, written.rejection);
}
+ match rejection {
+ Some(rejection) => Err(rejection.error),
+ None => Ok(()),
+ }
+ }
+
+ fn enable_at_least_once(&mut self) -> Result<(), Error> {
+ self.held = Some(HeldMessages::default());
Ok(())
}
}
#[cfg(test)]
mod tests {
- use super::route_value;
- use iggy_connector_sdk::Payload;
+ use super::*;
+ use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+ use iceberg::spec::{NestedField, PrimitiveType, Schema as IcebergSchema,
Type};
+ use iceberg::{CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
+ use std::time::Duration;
+
+ const ORIGIN: BatchOrigin<'static> = BatchOrigin {
+ connector_id: 1,
+ stream: "stream",
+ topic: "topic",
+ partition_id: 0,
+ };
+
+ async fn router(invalid_rows: InvalidRows, at_least_once: bool) ->
DynamicRouter {
+ let catalog = MemoryCatalogBuilder::default()
+ .load(
+ "memory",
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_owned(),
+ "memory://warehouse".to_owned(),
+ )]),
+ )
+ .await
+ .expect("Failed to create catalog");
+ let namespace = NamespaceIdent::new("test".to_owned());
+ catalog
+ .create_namespace(&namespace, HashMap::new())
+ .await
+ .expect("Failed to create namespace");
+ let schema = IcebergSchema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Long)).into(),
+ ])
+ .build()
+ .expect("Failed to build schema");
+ catalog
+ .create_table(
+ &namespace,
+ TableCreation::builder()
+ .name("events".to_owned())
+ .schema(schema)
+ .build(),
+ )
+ .await
+ .expect("Failed to create table");
+ let retry = RetryPolicy {
+ max_attempts: 1,
+ base_delay: Duration::from_millis(1),
+ max_delay: Duration::from_millis(1),
+ };
+ let mut router =
+ DynamicRouter::new(Box::new(catalog), "table".to_owned(),
invalid_rows, retry);
+ if at_least_once {
+ router
+ .enable_at_least_once()
+ .expect("the dynamic router supports at least once");
+ }
+ router
+ }
+
+ fn message(payload: simd_json::OwnedValue) -> ConsumedMessage {
+ ConsumedMessage {
+ id: 1,
+ offset: 7,
+ checksum: 0,
+ timestamp: 0,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Json(payload),
+ }
+ }
+
+ /// A message without the route field, one whose value names no namespace,
and one routed to
+ /// a table the catalog does not hold.
+ fn unroutable_messages() -> [ConsumedMessage; 3] {
+ [
+ message(simd_json::json!({ "id": 1 })),
+ message(simd_json::json!({ "table": 5 })),
+ message(simd_json::json!({ "table": "test.missing" })),
+ ]
+ }
+
+ #[tokio::test]
+ async fn
given_at_least_once_when_a_message_cannot_be_routed_should_fail_permanently() {
+ let router = router(InvalidRows::Fail, true).await;
+ for message in unroutable_messages() {
+ let error = router
+ .route_data(&ORIGIN, Schema::Json, vec![message])
+ .await
+ .expect_err("An unroutable message should fail the batch");
+
+ assert!(error.is_permanent(), "{error}");
+ assert!(error.to_string().contains("offset 7"), "{error}");
+ }
+ }
+
+ #[tokio::test]
+ async fn
given_invalid_rows_skip_when_a_message_cannot_be_routed_should_drop_it() {
+ for at_least_once in [true, false] {
+ let router = router(InvalidRows::Skip, at_least_once).await;
+
+ router
+ .route_data(&ORIGIN, Schema::Json,
unroutable_messages().into())
+ .await
+ .expect("Skipping should drop the unroutable messages");
+ }
+ }
+
+ #[tokio::test]
+ async fn
given_at_most_once_when_a_message_cannot_be_routed_should_drop_it() {
+ let router = router(InvalidRows::Fail, false).await;
+
+ router
+ .route_data(&ORIGIN, Schema::Json, unroutable_messages().into())
+ .await
+ .expect("At most once, an unroutable message is dropped");
+ }
#[test]
fn given_proto_text_holding_json_should_route_by_the_route_field() {
@@ -161,4 +349,67 @@ mod tests {
assert_eq!(route_value(&payload, "region"), None);
}
+
+ #[tokio::test]
+ async fn
given_at_least_once_when_a_message_cannot_be_routed_should_commit_the_messages_before_it()
+ {
+ let router = router(InvalidRows::Fail, true).await;
+ let routed = ConsumedMessage {
+ offset: 6,
+ ..message(simd_json::json!({ "table": "test.events", "id": 6 }))
+ };
+ let unroutable = message(simd_json::json!({ "id": 7 }));
+
+ let error = router
+ .route_data(&ORIGIN, Schema::Json, vec![routed, unroutable])
+ .await
+ .expect_err("The unroutable message should stop the batch");
+
+ assert!(error.to_string().contains("offset 7"), "{error}");
+ let table = router
+ .catalog
+ .load_table(&TableIdent::from_strs(["test", "events"]).expect("a
valid table name"))
+ .await
+ .expect("Failed to load table");
+ let snapshot = table
+ .metadata()
+ .current_snapshot()
+ .expect("The message before the unroutable one should be
committed");
+ assert_eq!(
+ snapshot
+ .summary()
+ .additional_properties
+ .get("added-records"),
+ Some(&"1".to_owned())
+ );
+ }
+
+ #[tokio::test]
+ async fn
given_a_catalog_that_refuses_the_load_when_skipping_rows_should_fail_without_skipping()
+ {
+ let catalog =
+
crate::router::commit::tests::rest_catalog_answering(http::StatusCode::FORBIDDEN).await;
+ let mut router = DynamicRouter::new(
+ Box::new(catalog),
+ "table".to_owned(),
+ InvalidRows::Skip,
+ RetryPolicy {
+ max_attempts: 1,
+ base_delay: Duration::from_millis(1),
+ max_delay: Duration::from_millis(1),
+ },
+ );
+ router
+ .enable_at_least_once()
+ .expect("the dynamic router supports at least once");
+ let routed = message(simd_json::json!({ "table": "test.events", "id":
7 }));
+
+ let error = router
+ .route_data(&ORIGIN, Schema::Json, vec![routed])
+ .await
+ .expect_err("A refused load says nothing about the message, so it
is not skipped");
+
+ assert!(error.is_permanent(), "{error}");
+ assert!(matches!(error, Error::InvalidConfigValue(_)), "{error}");
+ }
}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
index eb4e0acc6..de3865f08 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
@@ -18,6 +18,7 @@
use crate::InvalidRows;
use crate::router::arrow_streamer::JsonArrowReader;
use crate::router::commit::commit_data_files;
+use crate::router::redelivery::{HeldMessages, LastMessage,
snapshot_properties};
use crate::slice_user_table;
use arrow_array::RecordBatch;
use arrow_json::ReaderBuilder;
@@ -87,6 +88,7 @@ fn format_error_chain(err: &dyn std::error::Error) -> String {
mod arrow_streamer;
mod commit;
pub mod dynamic_router;
+mod redelivery;
pub mod static_router;
pub fn is_valid_namespaced_table(input: &str) -> bool {
@@ -94,43 +96,117 @@ pub fn is_valid_namespaced_table(input: &str) -> bool {
parts.len() >= 2 && parts.iter().all(|part| !part.is_empty())
}
-async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) ->
Option<Table> {
- let sliced_table = slice_user_table(route_field_val);
- let table_ident = TableIdent::from_strs(&sliced_table).ok()?;
+async fn load_table(route_field_val: &str, catalog: &dyn Catalog) ->
iceberg::Result<Table> {
+ let table_ident =
TableIdent::from_strs(slice_user_table(route_field_val))?;
+ catalog.load_table(&table_ident).await
+}
+
+/// A message that a table rejects with `invalid_rows = "fail"`. Every message
of the call before
+/// it is committed, so that storing its offset, as an operator does to move
past it, gives up
+/// this message alone.
+#[derive(Debug)]
+pub(crate) struct Rejection {
+ pub(crate) offset: u64,
+ pub(crate) error: Error,
+}
+
+/// What a write of a call's messages to one table did.
+#[derive(Debug, Default)]
+pub(crate) struct Written {
+ /// The rows committed, which leaves out the rows the sink skipped.
+ pub(crate) committed_rows: u64,
+ pub(crate) rejection: Option<Rejection>,
+}
+
+/// Rejections of several tables stop the call at the first rejected message.
+fn first_rejection(rejection: Option<Rejection>, other: Option<Rejection>) ->
Option<Rejection> {
+ match (rejection, other) {
+ (Some(rejection), Some(other)) if other.offset < rejection.offset =>
Some(other),
+ (rejection, other) => rejection.or(other),
+ }
+}
- catalog.load_table(&table_ident).await.ok()
+/// The messages before `rejection`, which a call still has to deliver.
+fn before<'m>(
+ messages: &'m [ConsumedMessage],
+ rejection: Option<&Rejection>,
+) -> &'m [ConsumedMessage] {
+ match rejection {
+ Some(rejection) => {
+ &messages[..messages.partition_point(|message| message.offset <
rejection.offset)]
+ }
+ None => messages,
+ }
}
-/// Returns the number of rows committed, which leaves out the rows the sink
skipped.
+/// Writes and commits `messages` to `table`. With `held`, the sink delivers
at least once: the
+/// messages that `table` already holds are not appended again, and the commit
records the last
+/// message it holds. `current` is a copy of `table` just loaded from the
catalog.
+#[allow(clippy::too_many_arguments)]
async fn write_data(
origin: &BatchOrigin<'_>,
messages: &[ConsumedMessage],
table: &Table,
+ current: Option<&Table>,
catalog: &dyn Catalog,
messages_schema: Schema,
invalid_rows: InvalidRows,
commit_retry: RetryPolicy,
-) -> Result<u64, Error> {
- let data_files =
- write_data_files(origin, messages, table, messages_schema,
invalid_rows).await?;
- if data_files.is_empty() {
- // Every row was skipped, and an empty append would still add a
snapshot.
- return Ok(0);
+ held: Option<&HeldMessages>,
+) -> Result<Written, Error> {
+ if messages.is_empty() {
+ return Ok(Written::default());
+ }
+ let messages = match held {
+ Some(held) => {
+ held.unheld(origin, table, current, catalog, messages)
+ .await?
+ }
+ None => messages,
+ };
+ if messages.is_empty() {
+ return Ok(Written::default());
}
+ let (data_files, rejection) =
+ write_data_files(origin, messages, table, messages_schema,
invalid_rows).await?;
+ let handled = before(messages, rejection.as_ref());
+ let (Some(last), false) = (handled.last(), data_files.is_empty()) else {
+ // Every row was skipped or rejected, and an empty append would still
add a snapshot.
+ return Ok(Written {
+ committed_rows: 0,
+ rejection,
+ });
+ };
let committed_rows = data_files.iter().map(DataFile::record_count).sum();
- commit_data_files(table, catalog, data_files, commit_retry).await?;
- Ok(committed_rows)
+ let recorded = held.map(|held| held.last_to_record(origin, table,
LastMessage::from(last)));
+ let properties = recorded
+ .map(|last| snapshot_properties(origin, last))
+ .unwrap_or_default();
+ let committed = commit_data_files(table, catalog, data_files, properties,
commit_retry).await;
+ if let (Some(held), Some(recorded)) = (held, recorded) {
+ match &committed {
+ Ok(()) => held.committed(origin, table, recorded),
+ Err(_) => held.forget(origin, table),
+ }
+ }
+ committed?;
+ Ok(Written {
+ committed_rows,
+ rejection,
+ })
}
/// Writes the JSON payloads as Parquet data files under the table location,
-/// laid out by the table's default partition spec. Nothing is committed here.
+/// laid out by the table's default partition spec. Nothing is committed here.
With
+/// `invalid_rows = "fail"`, the first message that the table cannot take ends
the write, and the
+/// data files hold the messages before it.
async fn write_data_files(
origin: &BatchOrigin<'_>,
messages: &[ConsumedMessage],
table: &Table,
messages_schema: Schema,
invalid_rows: InvalidRows,
-) -> Result<Vec<DataFile>, Error> {
+) -> Result<(Vec<DataFile>, Option<Rejection>), Error> {
let location =
DefaultLocationGenerator::new(table.metadata()).map_err(|err| {
error!(
"Failed to get location on table: {}. Error: {}",
@@ -161,28 +237,35 @@ async fn write_data_files(
let data_file_writer_builder =
DataFileWriterBuilder::new(rolling_file_writer_builder);
// Proto text holding JSON is the descriptor-less `proto_convert` fallback
- // and is written as the document it holds. Proto text that is not JSON is
- // skipped like any other payload without a document.
- let documents: Vec<(u64, Cow<'_, simd_json::OwnedValue>)> = messages
- .iter()
- .filter_map(|message| {
- let document = message.payload.json_document();
- if document.is_none() {
- warn!(
- "Unsupported type of payload, expected JSON, got {}",
- messages_schema.to_string()
- );
- }
- document.map(|document| (message.offset, document))
- })
- .collect();
+ // and is written as the document it holds. Proto text that is not JSON
holds
+ // no document, so it is an invalid row like any other payload without one.
+ let mut rejection = None;
+ let mut documents: Vec<(u64, Cow<'_, simd_json::OwnedValue>)> =
+ Vec::with_capacity(messages.len());
+ for message in messages {
+ if let Some(document) = message.payload.json_document() {
+ documents.push((message.offset, document));
+ continue;
+ }
+ let reason = format!(
+ "The message at offset: {} ({origin}) holds no JSON document for
table: {}, and the Iceberg sink requires schema = json, got {messages_schema}",
+ message.offset,
+ table.identifier()
+ );
+ if invalid_rows == InvalidRows::Skip {
+ warn!("Skipping a row. {reason}");
+ continue;
+ }
+ error!("{reason}");
+ rejection = Some(Rejection {
+ offset: message.offset,
+ error: Error::InvalidRecordValue(reason),
+ });
+ break;
+ }
if documents.is_empty() {
- error!(
- "Batch of {} messages has no JSON payloads, the Iceberg sink
requires schema = json",
- messages.len()
- );
- return Err(Error::InvalidPayloadType);
+ return Ok((Vec::new(), rejection));
}
let arrow_schema = Arc::new(
@@ -197,23 +280,30 @@ async fn write_data_files(
})?,
);
+ // The table's partition spec does not change between attempts, so a
writer that cannot be
+ // built fails the same way every time.
let mut writer = TableWriter::new(table,
data_file_writer_builder).map_err(|err| {
let chain = format_error_chain(&err);
error!("Error while constructing data file writer: {}", chain);
- Error::InitError(chain)
+ Error::SchemaMismatch(chain)
})?;
let write_result: Result<(), Error> = async {
for chunk in documents.chunks(ROWS_PER_BATCH) {
- if let Some(batch) = decode_rows(
+ let (batch, rejected) = decode_rows(
origin,
table.identifier(),
&arrow_schema,
chunk,
invalid_rows,
- )? {
+ )?;
+ if let Some(batch) = batch {
writer.write(batch).await?;
}
+ if rejected.is_some() {
+ rejection = first_rejection(rejection.take(), rejected);
+ break;
+ }
}
Ok(())
}
@@ -228,47 +318,59 @@ async fn write_data_files(
delete_uncommitted_files(table.file_io(), &data_files).await;
return Err(err);
}
- Ok(data_files)
+ Ok((data_files, rejection))
}
-/// Decodes the JSON documents into a record batch of `schema`. A row that
does not match fails
-/// the whole batch, unless `invalid_rows` skips it: then each such row is
logged with its offset,
-/// origin and table and left out, and the others are decoded. `None` when
every row is skipped.
+/// Decodes the JSON documents into a record batch of `schema`. A row that
does not match is
+/// logged with its offset, origin and table. `invalid_rows = "skip"` leaves
it out and decodes
+/// the others. `invalid_rows = "fail"` decodes the rows before it and rejects
it. The batch is
+/// `None` when no row is left.
fn decode_rows(
origin: &BatchOrigin<'_>,
table: &TableIdent,
schema: &SchemaRef,
documents: &[(u64, Cow<'_, simd_json::OwnedValue>)],
invalid_rows: InvalidRows,
-) -> Result<Option<RecordBatch>, Error> {
+) -> Result<(Option<RecordBatch>, Option<Rejection>), Error> {
let rows: Vec<&simd_json::OwnedValue> = documents
.iter()
.map(|(_, document)| document.as_ref())
.collect();
- let error = match read_batch(schema, &rows) {
- Ok(batch) => return Ok(batch),
- Err(error) => error,
- };
- if invalid_rows == InvalidRows::Fail {
- error!("Error while getting record batch: {error}");
- return Err(error);
+ if let Ok(batch) = read_batch(schema, &rows) {
+ return Ok((batch, None));
}
// The batch read only says that some row does not match, so each row is
read on its own.
let mut valid = Vec::with_capacity(rows.len());
+ let mut rejection = None;
for (offset, document) in documents {
- match read_batch(schema, &[document.as_ref()]) {
- Ok(_) => valid.push(document.as_ref()),
- Err(row_error) => warn!(
- "Skipping the row at offset: {offset} ({origin}) for table:
{table}, which does not match the table schema. {row_error}"
- ),
+ let row_error = match read_batch(schema, &[document.as_ref()]) {
+ Ok(_) => {
+ valid.push(document.as_ref());
+ continue;
+ }
+ Err(row_error) => row_error,
+ };
+ let reason = format!(
+ "The row at offset: {offset} ({origin}) does not match the schema
of table: {table}. {row_error}"
+ );
+ if invalid_rows == InvalidRows::Skip {
+ warn!("Skipping a row. {reason}");
+ continue;
}
+ error!("{reason}");
+ rejection = Some(Rejection {
+ offset: *offset,
+ error: Error::InvalidRecordValue(reason),
+ });
+ break;
}
if valid.is_empty() {
- return Ok(None);
+ return Ok((None, rejection));
}
- read_batch(schema, &valid).inspect_err(|error| {
+ let batch = read_batch(schema, &valid).inspect_err(|error| {
error!("Error while getting record batch: {error}");
- })
+ })?;
+ Ok((batch, rejection))
}
/// The reader allocates its buffers for a full batch, so the batch size is
the row count: a row
@@ -282,8 +384,9 @@ fn read_batch(
.with_batch_size(rows.len())
.build(JsonArrowReader::new(rows))
.map_err(|err| {
+ // The reader is built from the table schema alone, so it fails
the same way every time.
error!("Error while building Iceberg reader from message payload:
{err}");
- Error::InitError(err.to_string())
+ Error::SchemaMismatch(err.to_string())
})?;
reader
.next()
@@ -435,6 +538,11 @@ pub trait Router: std::fmt::Debug + Sync + Send {
messages_schema: Schema,
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error>;
+
+ /// From now on, fails a batch instead of dropping a message it cannot
route or a table that
+ /// does not load, and appends only the messages a table does not hold yet
when a batch is
+ /// delivered again. Refuses when the configuration cannot keep that
promise.
+ fn enable_at_least_once(&mut self) -> Result<(), Error>;
}
#[cfg(test)]
@@ -527,7 +635,7 @@ mod tests {
table: &Table,
payloads: &[Payload],
invalid_rows: InvalidRows,
- ) -> Result<Vec<DataFile>, Error> {
+ ) -> Result<(Vec<DataFile>, Option<Rejection>), Error> {
test_runtime().block_on(write_data_files(
&ORIGIN,
&messages(payloads),
@@ -538,7 +646,14 @@ mod tests {
}
fn write(table: &Table, payloads: &[Payload]) -> Vec<DataFile> {
- write_with(table, payloads, InvalidRows::Fail).expect("Failed to write
data files")
+ let (data_files, rejection) =
+ write_with(table, payloads, InvalidRows::Fail).expect("Failed to
write data files");
+ assert!(rejection.is_none(), "{rejection:?}");
+ data_files
+ }
+
+ fn rows(data_files: &[DataFile]) -> u64 {
+ data_files.iter().map(DataFile::record_count).sum()
}
/// Two rows that match the table and two that do not: a wrong type and a
missing required
@@ -574,11 +689,11 @@ mod tests {
}
#[test]
- fn
given_only_proto_text_that_is_not_json_should_fail_with_invalid_payload_type() {
+ fn
given_only_proto_text_that_is_not_json_when_failing_rows_should_reject_its_message()
{
let table = in_memory_table(UnboundPartitionSpec::builder().build());
let payloads = vec![Payload::Proto("id: 1".to_owned())];
- let error = test_runtime()
+ let (data_files, rejection) = test_runtime()
.block_on(write_data_files(
&ORIGIN,
&messages(&payloads),
@@ -586,9 +701,12 @@ mod tests {
Schema::Proto,
InvalidRows::Fail,
))
- .expect_err("proto text that is not JSON has no document to
write");
+ .expect("a rejected message is not a failed write");
- assert_eq!(error, Error::InvalidPayloadType);
+ assert!(data_files.is_empty());
+ let rejection = rejection.expect("proto text that is not JSON has no
document to write");
+ assert_eq!(rejection.offset, 0);
+ assert!(rejection.error.is_permanent(), "{}", rejection.error);
}
#[test]
@@ -660,30 +778,40 @@ mod tests {
}
#[test]
- fn
given_batch_without_json_payloads_should_fail_with_invalid_payload_type() {
+ fn
given_a_text_payload_when_skipping_rows_should_write_nothing_and_reject_nothing()
{
let table = in_memory_table(UnboundPartitionSpec::builder().build());
let payloads = vec![Payload::Text("not json".to_string())];
- let result = test_runtime().block_on(write_data_files(
- &ORIGIN,
- &messages(&payloads),
- &table,
- Schema::Text,
- InvalidRows::Fail,
- ));
+ let (data_files, rejection) = test_runtime()
+ .block_on(write_data_files(
+ &ORIGIN,
+ &messages(&payloads),
+ &table,
+ Schema::Text,
+ InvalidRows::Skip,
+ ))
+ .expect("skipping every row is not an error");
- assert!(matches!(result, Err(Error::InvalidPayloadType)));
+ assert!(data_files.is_empty());
+ assert!(rejection.is_none());
}
#[test]
- fn
given_rows_that_do_not_match_when_invalid_rows_fail_should_fail_the_batch() {
+ fn
given_rows_that_do_not_match_when_invalid_rows_fail_should_write_the_rows_before_the_first()
+ {
let table = in_memory_table(UnboundPartitionSpec::builder().build());
- let result = write_with(&table, &rows_with_two_invalid(),
InvalidRows::Fail);
+ let (data_files, rejection) =
+ write_with(&table, &rows_with_two_invalid(), InvalidRows::Fail)
+ .expect("a rejected row is not a failed write");
+ assert_eq!(rows(&data_files), 1);
+ let rejection = rejection.expect("the first row that does not match is
rejected");
+ assert_eq!(rejection.offset, 1);
assert!(
- matches!(result, Err(Error::InvalidRecordValue(_))),
- "got {result:?}"
+ matches!(rejection.error, Error::InvalidRecordValue(_)),
+ "got {:?}",
+ rejection.error
);
}
@@ -691,11 +819,12 @@ mod tests {
fn
given_rows_that_do_not_match_when_invalid_rows_skip_should_write_the_other_rows()
{
let table = in_memory_table(UnboundPartitionSpec::builder().build());
- let data_files = write_with(&table, &rows_with_two_invalid(),
InvalidRows::Skip)
- .expect("the matching rows should be written");
+ let (data_files, rejection) =
+ write_with(&table, &rows_with_two_invalid(), InvalidRows::Skip)
+ .expect("the matching rows should be written");
- let written: u64 = data_files.iter().map(DataFile::record_count).sum();
- assert_eq!(written, 2);
+ assert_eq!(rows(&data_files), 2);
+ assert!(rejection.is_none());
}
#[test]
@@ -705,10 +834,11 @@ mod tests {
simd_json::json!({ "id": "not a number", "region": "eu" }),
)];
- let data_files = write_with(&table, &payloads, InvalidRows::Skip)
+ let (data_files, rejection) = write_with(&table, &payloads,
InvalidRows::Skip)
.expect("skipping every row is not an error");
assert!(data_files.is_empty());
+ assert!(rejection.is_none());
}
#[test]
@@ -717,24 +847,26 @@ mod tests {
let mut payloads = json_payloads(&[(1, "eu"); ROWS_PER_BATCH + 10]);
payloads[ROWS_PER_BATCH + 5] = Payload::Json(simd_json::json!({ "id":
"not a number" }));
- let data_files = write_with(&table, &payloads, InvalidRows::Skip)
+ let (data_files, _) = write_with(&table, &payloads, InvalidRows::Skip)
.expect("the matching rows should be written");
- let written: u64 = data_files.iter().map(DataFile::record_count).sum();
- assert_eq!(written, (ROWS_PER_BATCH + 9) as u64);
+ assert_eq!(rows(&data_files), (ROWS_PER_BATCH + 9) as u64);
}
#[test]
- fn
given_an_invalid_row_in_a_later_batch_when_invalid_rows_fail_should_fail_the_batch()
{
+ fn
given_an_invalid_row_in_a_later_batch_when_invalid_rows_fail_should_write_the_rows_before_it()
+ {
let table = in_memory_table(UnboundPartitionSpec::builder().build());
let mut payloads = json_payloads(&[(1, "eu"); ROWS_PER_BATCH + 10]);
payloads[ROWS_PER_BATCH + 5] = Payload::Json(simd_json::json!({ "id":
"not a number" }));
- let result = write_with(&table, &payloads, InvalidRows::Fail);
+ let (data_files, rejection) = write_with(&table, &payloads,
InvalidRows::Fail)
+ .expect("a rejected row is not a failed write");
- assert!(
- matches!(result, Err(Error::InvalidRecordValue(_))),
- "got {result:?}"
+ assert_eq!(rows(&data_files), (ROWS_PER_BATCH + 5) as u64);
+ assert_eq!(
+ rejection.map(|rejection| rejection.offset),
+ Some((ROWS_PER_BATCH + 5) as u64)
);
}
}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/redelivery.rs
b/core/connectors/sinks/iceberg_sink/src/router/redelivery.rs
new file mode 100644
index 000000000..f90618ea6
--- /dev/null
+++ b/core/connectors/sinks/iceberg_sink/src/router/redelivery.rs
@@ -0,0 +1,385 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Which messages of a partition a table already holds. At least once, the
runtime hands a
+//! message over again when it cannot prove it delivered, for example after a
crash between the
+//! commit and the offset store, or after a batch committed to one table and
failed on another.
+//! Every append records the last message it holds in its snapshot summary,
and a delivery of
+//! the same messages again appends only the ones after the newest append.
+
+use crate::router::BatchOrigin;
+use crate::router::commit::table_load_error;
+use dashmap::DashMap;
+use iceberg::spec::Snapshot;
+use iceberg::table::Table;
+use iceberg::util::snapshot::ancestors_of;
+use iceberg::{Catalog, TableIdent};
+use iggy_connector_sdk::{ConsumedMessage, Error};
+use std::collections::{BTreeMap, HashMap};
+use tracing::{info, warn};
+use uuid::Uuid;
+
+const STREAM_PROPERTY: &str = "iggy.stream";
+const TOPIC_PROPERTY: &str = "iggy.topic";
+const PARTITION_ID_PROPERTY: &str = "iggy.partition-id";
+const LAST_OFFSET_PROPERTY: &str = "iggy.last-offset";
+const LAST_MESSAGE_ID_PROPERTY: &str = "iggy.last-message-id";
+const LAST_TIMESTAMP_PROPERTY: &str = "iggy.last-timestamp";
+
+/// The last message of a partition that an append holds. The message ID and
the time the
+/// server stored the message tell it apart from another message at the same
offset in a later
+/// history of the partition.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct LastMessage {
+ offset: u64,
+ id: u128,
+ timestamp: u64,
+}
+
+impl From<&ConsumedMessage> for LastMessage {
+ fn from(message: &ConsumedMessage) -> Self {
+ Self {
+ offset: message.offset,
+ id: message.id,
+ timestamp: message.timestamp,
+ }
+ }
+}
+
+impl LastMessage {
+ fn from_properties(properties: &HashMap<String, String>) -> Option<Self> {
+ Some(Self {
+ offset: properties.get(LAST_OFFSET_PROPERTY)?.parse().ok()?,
+ id: properties.get(LAST_MESSAGE_ID_PROPERTY)?.parse().ok()?,
+ timestamp: properties.get(LAST_TIMESTAMP_PROPERTY)?.parse().ok()?,
+ })
+ }
+}
+
+/// The snapshot summary properties that record `last` as the last message of
`origin` the
+/// append holds.
+pub(crate) fn snapshot_properties(
+ origin: &BatchOrigin<'_>,
+ last: LastMessage,
+) -> HashMap<String, String> {
+ HashMap::from([
+ (STREAM_PROPERTY.to_owned(), origin.stream.to_owned()),
+ (TOPIC_PROPERTY.to_owned(), origin.topic.to_owned()),
+ (
+ PARTITION_ID_PROPERTY.to_owned(),
+ origin.partition_id.to_string(),
+ ),
+ (LAST_OFFSET_PROPERTY.to_owned(), last.offset.to_string()),
+ (LAST_MESSAGE_ID_PROPERTY.to_owned(), last.id.to_string()),
+ (
+ LAST_TIMESTAMP_PROPERTY.to_owned(),
+ last.timestamp.to_string(),
+ ),
+ ])
+}
+
+/// A partition delivered to a table. The table UUID tells a table recreated
under the same
+/// name apart.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct TablePartition {
+ table: Uuid,
+ stream: String,
+ topic: String,
+ partition_id: u32,
+}
+
+impl TablePartition {
+ fn new(origin: &BatchOrigin<'_>, table: &Table) -> Self {
+ Self {
+ table: table.metadata().uuid(),
+ stream: origin.stream.to_owned(),
+ topic: origin.topic.to_owned(),
+ partition_id: origin.partition_id,
+ }
+ }
+}
+
+/// The appends of one partition that a table holds.
+#[derive(Debug, Default)]
+struct Appends {
+ /// The last message of each append, by offset. Where two histories of the
partition
+ /// appended at the same offset, the newer one is kept.
+ last_messages: BTreeMap<u64, LastMessage>,
+ newest: Option<LastMessage>,
+}
+
+impl Appends {
+ fn record(&mut self, last: LastMessage) {
+ self.last_messages.insert(last.offset, last);
+ self.newest = Some(last);
+ }
+}
+
+/// What the sink knows each table holds of each partition it delivers. A
partition is unknown
+/// until the sink reads the table's snapshot history, and again after a
commit that failed,
+/// since the catalog may have applied it.
+#[derive(Debug, Default)]
+pub(crate) struct HeldMessages {
+ known: DashMap<TablePartition, Appends>,
+}
+
+impl HeldMessages {
+ /// The tail of `messages` that `table` does not hold yet. `current` is a
copy of `table`
+ /// just loaded from the catalog, which saves loading it again.
+ pub(crate) async fn unheld<'m>(
+ &self,
+ origin: &BatchOrigin<'_>,
+ table: &Table,
+ current: Option<&Table>,
+ catalog: &dyn Catalog,
+ messages: &'m [ConsumedMessage],
+ ) -> Result<&'m [ConsumedMessage], Error> {
+ let key = TablePartition::new(origin, table);
+ if !self.known.contains_key(&key) {
+ let appends = match current {
+ Some(current) => appends_in(current, origin),
+ None => {
+ let identifier = table.identifier();
+ let current = catalog
+ .load_table(identifier)
+ .await
+ .map_err(|error|
table_load_error(&identifier.to_string(), &error))?;
+ appends_in(¤t, origin)
+ }
+ };
+ self.known.entry(key.clone()).or_insert(appends);
+ }
+ let held = match self.known.get_mut(&key) {
+ Some(mut appends) => {
+ match held_prefix(origin, table.identifier(), messages,
&appends) {
+ Prefix::Held(held) => held,
+ Prefix::NewHistory => {
+ *appends = Appends::default();
+ 0
+ }
+ Prefix::Unproven => 0,
+ }
+ }
+ None => 0,
+ };
+ Ok(&messages[held..])
+ }
+
+ /// The last message of a commit to record in its snapshot. A batch that
the table could not
+ /// prove it held, but that ends below the newest append, records that
append again, so that
+ /// the record never moves back within one history of the partition.
+ pub(crate) fn last_to_record(
+ &self,
+ origin: &BatchOrigin<'_>,
+ table: &Table,
+ last: LastMessage,
+ ) -> LastMessage {
+ self.known
+ .get(&TablePartition::new(origin, table))
+ .and_then(|appends| appends.newest)
+ .filter(|newest| newest.offset > last.offset)
+ .unwrap_or(last)
+ }
+
+ pub(crate) fn committed(&self, origin: &BatchOrigin<'_>, table: &Table,
last: LastMessage) {
+ self.known
+ .entry(TablePartition::new(origin, table))
+ .or_default()
+ .record(last);
+ }
+
+ pub(crate) fn forget(&self, origin: &BatchOrigin<'_>, table: &Table) {
+ self.known.remove(&TablePartition::new(origin, table));
+ }
+}
+
+/// The appends of `origin` on the current snapshot's ancestry. Snapshots that
a rollback or a
+/// replace left off that ancestry do not count, since the table no longer
holds their rows.
+fn appends_in(table: &Table, origin: &BatchOrigin<'_>) -> Appends {
+ let mut appends = Appends::default();
+ let Some(current) = table.metadata().current_snapshot_id() else {
+ return appends;
+ };
+ let partition_id = origin.partition_id.to_string();
+ for snapshot in ancestors_of(&table.metadata_ref(), current) {
+ if !records(&snapshot, origin, &partition_id) {
+ continue;
+ }
+ let properties = &snapshot.summary().additional_properties;
+ let Some(last) = LastMessage::from_properties(properties) else {
+ warn!(
+ "Iceberg table {} has snapshot {} of {origin} without a
readable last message. Skipping it",
+ table.identifier(),
+ snapshot.snapshot_id()
+ );
+ continue;
+ };
+ // The walk runs from the newest snapshot back, so the first one seen
at an offset wins.
+ appends.last_messages.entry(last.offset).or_insert(last);
+ appends.newest.get_or_insert(last);
+ }
+ appends
+}
+
+fn records(snapshot: &Snapshot, origin: &BatchOrigin<'_>, partition_id: &str)
-> bool {
+ let properties = &snapshot.summary().additional_properties;
+ properties.get(STREAM_PROPERTY).map(String::as_str) == Some(origin.stream)
+ && properties.get(TOPIC_PROPERTY).map(String::as_str) ==
Some(origin.topic)
+ && properties.get(PARTITION_ID_PROPERTY).map(String::as_str) ==
Some(partition_id)
+}
+
+/// What a table holds of the start of a batch.
+#[derive(Debug, PartialEq, Eq)]
+enum Prefix {
+ /// This many messages, proven to be the history the table holds.
+ Held(usize),
+ /// None: the table holds other messages at the batch's offsets.
+ NewHistory,
+ /// None that can be proven, although the batch reaches offsets the table
holds.
+ Unproven,
+}
+
+/// What the table holds of the start of `messages`. Messages count as held
only when the batch
+/// carries the last message of one of the table's appends, which proves that
it is the history
+/// the table holds. Then every message up to the newest append is held.
+fn held_prefix(
+ origin: &BatchOrigin<'_>,
+ table: &TableIdent,
+ messages: &[ConsumedMessage],
+ appends: &Appends,
+) -> Prefix {
+ let Some(newest) = appends.newest else {
+ return Prefix::Held(0);
+ };
+ // A run of one partition carries ascending offsets.
+ let held = messages.partition_point(|message| message.offset <=
newest.offset);
+ if held == 0 {
+ return Prefix::Held(0);
+ }
+ let mut other_history = false;
+ for message in &messages[..held] {
+ match appends.last_messages.get(&message.offset) {
+ Some(last) if *last == LastMessage::from(message) => {
+ info!(
+ "Skipping {held} messages from offset: {} to offset: {}
({origin}) that Iceberg table {table} already holds",
+ messages[0].offset,
+ messages[held - 1].offset,
+ );
+ return Prefix::Held(held);
+ }
+ Some(_) => other_history = true,
+ None => {}
+ }
+ }
+ if other_history {
+ warn!(
+ "Iceberg table {table} holds other messages at the offsets of the
batch ({origin}), so the partition has a new history. Appending every message
of the batch"
+ );
+ return Prefix::NewHistory;
+ }
+ warn!(
+ "Iceberg table {table} holds messages up to offset: {} ({origin}), but
the batch carries no message that proves which ones. Appending every message of
the batch, which can append some twice",
+ newest.offset
+ );
+ Prefix::Unproven
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_connector_sdk::Payload;
+
+ const ORIGIN: BatchOrigin<'static> = BatchOrigin {
+ connector_id: 1,
+ stream: "stream",
+ topic: "topic",
+ partition_id: 2,
+ };
+
+ fn table() -> TableIdent {
+ TableIdent::from_strs(["ns", "table"]).expect("a valid table name")
+ }
+
+ fn message(offset: u64) -> ConsumedMessage {
+ ConsumedMessage {
+ id: u128::from(offset) + 100,
+ offset,
+ checksum: 0,
+ timestamp: offset + 1_000,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Raw(Vec::new()),
+ }
+ }
+
+ fn messages(offsets: std::ops::RangeInclusive<u64>) ->
Vec<ConsumedMessage> {
+ offsets.map(message).collect()
+ }
+
+ fn appends(last_offsets: &[u64]) -> Appends {
+ let mut appends = Appends::default();
+ for &offset in last_offsets {
+ appends.record(LastMessage::from(&message(offset)));
+ }
+ appends
+ }
+
+ #[test]
+ fn given_nothing_held_should_append_every_message() {
+ let prefix = held_prefix(&ORIGIN, &table(), &messages(0..=1),
&Appends::default());
+
+ assert_eq!(prefix, Prefix::Held(0));
+ }
+
+ #[test]
+ fn given_a_batch_carrying_the_newest_append_should_hold_up_to_it() {
+ let prefix = held_prefix(&ORIGIN, &table(), &messages(3..=5),
&appends(&[4]));
+
+ assert_eq!(prefix, Prefix::Held(2));
+ }
+
+ #[test]
+ fn
given_a_backlog_batch_carrying_an_older_append_should_hold_up_to_the_newest() {
+ let prefix = held_prefix(&ORIGIN, &table(), &messages(0..=1),
&appends(&[1, 3, 5]));
+
+ assert_eq!(prefix, Prefix::Held(2));
+ }
+
+ #[test]
+ fn
given_another_message_at_an_appended_offset_should_report_a_new_history() {
+ let mut batch = messages(3..=4);
+ batch[1].timestamp += 1;
+
+ let prefix = held_prefix(&ORIGIN, &table(), &batch, &appends(&[4]));
+
+ assert_eq!(prefix, Prefix::NewHistory);
+ }
+
+ #[test]
+ fn given_a_batch_after_the_newest_append_should_append_every_message() {
+ let prefix = held_prefix(&ORIGIN, &table(), &messages(5..=6),
&appends(&[4]));
+
+ assert_eq!(prefix, Prefix::Held(0));
+ }
+
+ #[test]
+ fn given_a_batch_without_an_appended_message_should_append_every_message()
{
+ let prefix = held_prefix(&ORIGIN, &table(), &messages(2..=3),
&appends(&[1, 4]));
+
+ assert_eq!(prefix, Prefix::Unproven);
+ }
+}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/static_router.rs
b/core/connectors/sinks/iceberg_sink/src/router/static_router.rs
index ec2376ece..7d3d8cc58 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/static_router.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/static_router.rs
@@ -16,20 +16,49 @@
// under the License.
use crate::InvalidRows;
-use crate::router::{BatchOrigin, Router, is_valid_namespaced_table,
table_exists, write_data};
+use crate::router::commit::table_load_error;
+use crate::router::redelivery::HeldMessages;
+use crate::router::{
+ BatchOrigin, Router, before, first_rejection, format_error_chain,
is_valid_namespaced_table,
+ load_table, write_data,
+};
use async_trait::async_trait;
use iceberg::Catalog;
use iceberg::table::Table;
use iggy_connector_sdk::retry::RetryPolicy;
use iggy_connector_sdk::{ConsumedMessage, Error, Schema};
+use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use tracing::{error, info, warn};
+/// A table that the configuration declares. A table that does not load is
loaded again by the
+/// next batch, so that a catalog that was unreachable at startup does not
drop it for good.
+#[derive(Debug)]
+struct DeclaredTable {
+ name: String,
+ table: Mutex<Option<Arc<Table>>>,
+}
+
+impl DeclaredTable {
+ fn loaded(&self) -> Option<Arc<Table>> {
+ self.lock().clone()
+ }
+
+ fn lock(&self) -> MutexGuard<'_, Option<Arc<Table>>> {
+ // The guarded value is replaced whole, so a panic while the lock was
held cannot leave it
+ // half written.
+ self.table.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
#[derive(Debug)]
pub(crate) struct StaticRouter {
- tables: Vec<Table>,
+ tables: Vec<DeclaredTable>,
+ /// Declared names that name no table, which no load can fix.
+ invalid_names: Vec<String>,
catalog: Box<dyn Catalog>,
invalid_rows: InvalidRows,
commit_retry: RetryPolicy,
+ held: Option<HeldMessages>,
}
impl StaticRouter {
@@ -39,44 +68,80 @@ impl StaticRouter {
invalid_rows: InvalidRows,
commit_retry: RetryPolicy,
) -> Result<Self, Error> {
- let mut tables: Vec<Table> = Vec::with_capacity(declared_tables.len());
+ let mut tables = Vec::with_capacity(declared_tables.len());
+ let mut invalid_names = Vec::new();
for declared_table in declared_tables {
if !is_valid_namespaced_table(declared_table) {
error!(
"Declared table {} is not valid. It has to include at
least one namespace before the table name separated by '.' character",
declared_table
);
+ invalid_names.push(declared_table.clone());
continue;
}
- let table = match table_exists(declared_table,
catalog.as_ref()).await {
- Some(table) => table,
- None => {
+ let table = match load_table(declared_table,
catalog.as_ref()).await {
+ Ok(table) => Some(Arc::new(table)),
+ Err(err) => {
warn!(
- "Declared table {} doesn't exist in the configured
catalog. Skipping...",
- declared_table
+ "Declared table {declared_table} does not load from
the configured catalog, so each batch tries again. {}",
+ format_error_chain(&err)
);
- continue;
+ None
}
};
- tables.push(table);
+ tables.push(DeclaredTable {
+ name: declared_table.clone(),
+ table: Mutex::new(table),
+ });
}
+ let loaded = tables
+ .iter()
+ .filter(|declared| declared.loaded().is_some())
+ .count();
info!(
- "Static router found {} tables on iceberg catalog from {} tables
declared",
- tables.len(),
+ "Static router found {loaded} tables on iceberg catalog from {}
tables declared",
declared_tables.len()
);
- if tables.is_empty() {
+ if loaded == 0 {
error!("No valid tables found. Can't initiate Iceberg connector");
return Err(Error::InvalidConfig);
}
Ok(StaticRouter {
tables,
+ invalid_names,
catalog,
invalid_rows,
commit_retry,
+ held: None,
})
}
+
+ /// The table of `declared`, loaded from the catalog when no earlier batch
could. At most
+ /// once, a table that does not load is skipped for this batch. At least
once, it fails the
+ /// batch, so that its messages are delivered again instead of lost.
+ async fn table(&self, declared: &DeclaredTable) ->
Result<Option<Arc<Table>>, Error> {
+ if let Some(table) = declared.loaded() {
+ return Ok(Some(table));
+ }
+ match load_table(&declared.name, self.catalog.as_ref()).await {
+ Ok(table) => {
+ let table = Arc::new(table);
+ info!("Declared table {} loaded from the catalog",
declared.name);
+ *declared.lock() = Some(Arc::clone(&table));
+ Ok(Some(table))
+ }
+ Err(err) if self.held.is_some() =>
Err(table_load_error(&declared.name, &err)),
+ Err(err) => {
+ warn!(
+ "Skipping declared table {} for this batch, it does not
load from the catalog. {}",
+ declared.name,
+ format_error_chain(&err)
+ );
+ Ok(None)
+ }
+ }
+ }
}
#[async_trait]
@@ -87,24 +152,175 @@ impl Router for StaticRouter {
messages_schema: Schema,
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error> {
- for table in &self.tables {
- let committed_rows = write_data(
+ let mut rejection = None;
+ for declared in &self.tables {
+ let Some(table) = self.table(declared).await? else {
+ continue;
+ };
+ // Once a table rejects a message, the others still need every
message before it.
+ let messages = before(&messages, rejection.as_ref());
+ let written = write_data(
origin,
- &messages,
- table,
+ messages,
+ &table,
+ None,
self.catalog.as_ref(),
messages_schema,
self.invalid_rows,
self.commit_retry,
+ self.held.as_ref(),
)
.await?;
info!(
- "Routed {} input rows to iceberg table {}, committed
{committed_rows}",
+ "Routed {} input rows to iceberg table {}, committed {}",
messages.len(),
- table.identifier().name()
+ table.identifier().name(),
+ written.committed_rows
);
+ rejection = first_rejection(rejection, written.rejection);
}
+ match rejection {
+ Some(rejection) => Err(rejection.error),
+ None => Ok(()),
+ }
+ }
+
+ fn enable_at_least_once(&mut self) -> Result<(), Error> {
+ if !self.invalid_names.is_empty() {
+ return Err(Error::Unsupported(format!(
+ "declared tables {:?} are not valid table names, so their
messages could never be delivered. Fix or remove them in `tables`",
+ self.invalid_names
+ )));
+ }
+ self.held = Some(HeldMessages::default());
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+ use iceberg::spec::{NestedField, PrimitiveType, Schema as IcebergSchema,
Type};
+ use iceberg::{CatalogBuilder, NamespaceIdent, TableCreation, TableIdent};
+ use iggy_connector_sdk::Payload;
+ use std::collections::HashMap;
+ use std::time::Duration;
+
+ const ORIGIN: BatchOrigin<'static> = BatchOrigin {
+ connector_id: 1,
+ stream: "stream",
+ topic: "topic",
+ partition_id: 0,
+ };
+
+ /// A router over `declared` tables, of which the catalog holds only
`test.events`.
+ async fn router(declared: &[&str]) -> StaticRouter {
+ let catalog = MemoryCatalogBuilder::default()
+ .load(
+ "memory",
+ HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_owned(),
+ "memory://warehouse".to_owned(),
+ )]),
+ )
+ .await
+ .expect("Failed to create catalog");
+ let namespace = NamespaceIdent::new("test".to_owned());
+ catalog
+ .create_namespace(&namespace, HashMap::new())
+ .await
+ .expect("Failed to create namespace");
+ let schema = IcebergSchema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Long)).into(),
+ ])
+ .build()
+ .expect("Failed to build schema");
+ catalog
+ .create_table(
+ &namespace,
+ TableCreation::builder()
+ .name("events".to_owned())
+ .schema(schema)
+ .build(),
+ )
+ .await
+ .expect("Failed to create table");
+ let declared: Vec<String> = declared.iter().map(|name|
(*name).to_owned()).collect();
+ let retry = RetryPolicy {
+ max_attempts: 1,
+ base_delay: Duration::from_millis(1),
+ max_delay: Duration::from_millis(1),
+ };
+ StaticRouter::new(Box::new(catalog), &declared, InvalidRows::Fail,
retry)
+ .await
+ .expect("One declared table should load")
+ }
+
+ fn messages() -> Vec<ConsumedMessage> {
+ vec![ConsumedMessage {
+ id: 1,
+ offset: 0,
+ checksum: 0,
+ timestamp: 1,
+ origin_timestamp: 0,
+ headers: None,
+ payload: Payload::Json(simd_json::json!({ "id": 1 })),
+ }]
+ }
+
+ async fn snapshots(router: &StaticRouter) -> usize {
+ router
+ .catalog
+ .load_table(&TableIdent::from_strs(["test", "events"]).expect("a
valid table name"))
+ .await
+ .expect("Failed to load table")
+ .metadata()
+ .snapshots()
+ .count()
+ }
+
+ #[tokio::test]
+ async fn
given_a_declared_table_that_does_not_load_when_at_most_once_should_skip_it() {
+ let router = router(&["test.events", "test.later"]).await;
+
+ router
+ .route_data(&ORIGIN, Schema::Json, messages())
+ .await
+ .expect("At most once, a table that does not load is skipped");
+
+ assert_eq!(snapshots(&router).await, 1);
+ }
+
+ #[tokio::test]
+ async fn
given_a_declared_table_that_does_not_load_when_at_least_once_should_fail_the_batch()
{
+ let mut router = router(&["test.events", "test.later"]).await;
+ router
+ .enable_at_least_once()
+ .expect("Every declared name is valid");
+
+ let error = router
+ .route_data(&ORIGIN, Schema::Json, messages())
+ .await
+ .expect_err("At least once, a table that does not load fails the
batch");
+
+ assert!(
+ error.is_permanent(),
+ "a missing table stays missing: {error}"
+ );
+ assert!(error.to_string().contains("test.later"), "{error}");
+ }
+
+ #[tokio::test]
+ async fn
given_a_declared_name_that_names_no_table_should_refuse_at_least_once() {
+ let mut router = router(&["test.events", "events"]).await;
+
+ let error = router
+ .enable_at_least_once()
+ .expect_err("A name without a namespace can never be written");
+
+ assert!(matches!(error, Error::Unsupported(_)), "{error}");
+ }
+}
diff --git a/core/connectors/sinks/iceberg_sink/src/sink.rs
b/core/connectors/sinks/iceberg_sink/src/sink.rs
index 7418502f5..8530a8096 100644
--- a/core/connectors/sinks/iceberg_sink/src/sink.rs
+++ b/core/connectors/sinks/iceberg_sink/src/sink.rs
@@ -130,6 +130,15 @@ impl Sink for IcebergSink {
Ok(())
}
+ /// The sink commits every batch inside `consume`, so the default flush
has nothing left to
+ /// prove.
+ fn enable_at_least_once(&mut self) -> Result<(), Error> {
+ let Some(router) = self.router.as_mut() else {
+ return Err(Error::InvalidState);
+ };
+ router.enable_at_least_once()
+ }
+
async fn close(&mut self) -> Result<(), Error> {
info!("Iceberg sink connector with ID: {} is closed.", self.id);
Ok(())
diff --git a/core/integration/src/harness/handle/connectors_runtime.rs
b/core/integration/src/harness/handle/connectors_runtime.rs
index a15fe7eb3..d08b942de 100644
--- a/core/integration/src/harness/handle/connectors_runtime.rs
+++ b/core/integration/src/harness/handle/connectors_runtime.rs
@@ -89,6 +89,17 @@ impl ConnectorsRuntimeHandle {
self.iggy_connection_options = None;
}
+ /// Kills the runtime with SIGKILL, so that no flush or offset store of a
graceful stop
+ /// runs. Start it again with [`TestBinary::start`] and
[`IggyServerDependent::wait_ready`].
+ pub fn kill(&mut self) -> Result<(), TestBinaryError> {
+ let Some(mut child) = self.child_handle.take() else {
+ return Ok(());
+ };
+ child.kill()?;
+ child.wait()?;
+ Ok(())
+ }
+
fn build_envs(&mut self) {
let state_path =
self.context.connectors_runtime_state_path(self.server_id);
self.envs.insert(
@@ -180,13 +191,14 @@ impl TestBinary for ConnectorsRuntimeHandle {
let stdout_path =
self.context.connectors_runtime_stdout_path(self.server_id);
let stderr_path =
self.context.connectors_runtime_stderr_path(self.server_id);
+ // Appended, so that a start after a kill keeps the logs of the
killed run.
let stdout_file =
- File::create(&stdout_path).map_err(|e|
TestBinaryError::FileSystemError {
+ append_to(&stdout_path).map_err(|e|
TestBinaryError::FileSystemError {
path: stdout_path.clone(),
source: e,
})?;
let stderr_file =
- File::create(&stderr_path).map_err(|e|
TestBinaryError::FileSystemError {
+ append_to(&stderr_path).map_err(|e|
TestBinaryError::FileSystemError {
path: stderr_path.clone(),
source: e,
})?;
@@ -322,3 +334,7 @@ impl Drop for ConnectorsRuntimeHandle {
common::dump_logs_on_panic("Iggy connectors", &self.stdout_path,
&self.stderr_path);
}
}
+
+fn append_to(path: &std::path::Path) -> std::io::Result<File> {
+ OpenOptions::new().create(true).append(true).open(path)
+}
diff --git a/core/integration/tests/connectors/fixtures/iceberg/container.rs
b/core/integration/tests/connectors/fixtures/iceberg/container.rs
index 6736db26f..00e58e7d3 100644
--- a/core/integration/tests/connectors/fixtures/iceberg/container.rs
+++ b/core/integration/tests/connectors/fixtures/iceberg/container.rs
@@ -18,7 +18,7 @@
use std::collections::HashMap;
use async_trait::async_trait;
-use integration::harness::{SinglePortReserver, TestBinaryError, TestFixture};
+use integration::harness::{SinglePortReserver, TestBinaryError, TestFixture,
seeds};
use reqwest_middleware::ClientWithMiddleware as HttpClient;
use reqwest_retry::RetryTransientMiddleware;
use reqwest_retry::policies::ExponentialBackoff;
@@ -54,6 +54,23 @@ pub const ENV_SINK_MAX_RETRIES: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONF
pub const ENV_SINK_RETRY_DELAY: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONFIG_RETRY_DELAY";
pub const ENV_SINK_MAX_RETRY_DELAY: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONFIG_MAX_RETRY_DELAY";
+pub const ENV_SINK_DELIVERY_GUARANTEE: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_DELIVERY_GUARANTEE";
+pub const ENV_SINK_DELIVERY_FLUSH_INTERVAL: &str =
+ "IGGY_CONNECTORS_SINK_ICEBERG_DELIVERY_FLUSH_INTERVAL";
+pub const ENV_SINK_DELIVERY_RETRY_INITIAL_BACKOFF: &str =
+ "IGGY_CONNECTORS_SINK_ICEBERG_DELIVERY_RETRY_INITIAL_BACKOFF";
+pub const ENV_SINK_DELIVERY_RETRY_MAX_BACKOFF: &str =
+ "IGGY_CONNECTORS_SINK_ICEBERG_DELIVERY_RETRY_MAX_BACKOFF";
+// Setting one field of a `[[streams]]` entry from the environment replaces
the whole entry.
+pub const ENV_SINK_STREAMS_0_STREAM: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_STREAMS_0_STREAM";
+pub const ENV_SINK_STREAMS_0_TOPICS: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_STREAMS_0_TOPICS";
+pub const ENV_SINK_STREAMS_0_SCHEMA: &str =
"IGGY_CONNECTORS_SINK_ICEBERG_STREAMS_0_SCHEMA";
+pub const ENV_SINK_STREAMS_0_BATCH_LENGTH: &str =
+ "IGGY_CONNECTORS_SINK_ICEBERG_STREAMS_0_BATCH_LENGTH";
+pub const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str =
+ "IGGY_CONNECTORS_SINK_ICEBERG_STREAMS_0_CONSUMER_GROUP";
+/// The messages of one batch of `IcebergDeferredCheckpointFixture`.
+pub const DEFERRED_CHECKPOINT_BATCH_LENGTH: u32 = 2;
pub const ENV_AWS_ACCESS_KEY_ID: &str = "AWS_ACCESS_KEY_ID";
pub const ENV_AWS_SECRET_ACCESS_KEY: &str = "AWS_SECRET_ACCESS_KEY";
@@ -356,6 +373,9 @@ pub struct SnapshotSummary {
pub added_records: Option<String>,
#[serde(rename = "changed-partition-count")]
pub changed_partition_count: Option<String>,
+ /// The last message that an at-least-once append holds.
+ #[serde(rename = "iggy.last-offset")]
+ pub last_offset: Option<String>,
}
impl IcebergFixture {
@@ -553,6 +573,123 @@ impl TestFixture for IcebergCatalogRestartFixture {
}
}
+/// Same table as `IcebergPreCreatedFixture`, with a sink that delivers at
least once. Its own
+/// commit retries run out within a second, so a longer catalog outage reaches
the runtime, which
+/// hands the batch over again.
+pub struct IcebergAtLeastOnceFixture {
+ inner: IcebergPreCreatedFixture,
+}
+
+impl std::ops::Deref for IcebergAtLeastOnceFixture {
+ type Target = IcebergFixture;
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+impl IcebergOps for IcebergAtLeastOnceFixture {
+ fn catalog_url(&self) -> &str {
+ self.inner.catalog_url()
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ self.inner.http_client()
+ }
+}
+
+#[async_trait]
+impl TestFixture for IcebergAtLeastOnceFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ Ok(Self {
+ inner: IcebergPreCreatedFixture::setup().await?,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.inner.connectors_runtime_envs();
+ envs.insert(
+ ENV_SINK_DELIVERY_GUARANTEE.to_string(),
+ "at_least_once".to_string(),
+ );
+ envs.insert(ENV_SINK_MAX_RETRIES.to_string(), "2".to_string());
+ envs.insert(ENV_SINK_RETRY_DELAY.to_string(), "200ms".to_string());
+ envs.insert(ENV_SINK_MAX_RETRY_DELAY.to_string(), "200ms".to_string());
+ envs.insert(
+ ENV_SINK_DELIVERY_RETRY_INITIAL_BACKOFF.to_string(),
+ "200ms".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_DELIVERY_RETRY_MAX_BACKOFF.to_string(),
+ "1s".to_string(),
+ );
+ envs
+ }
+}
+
+/// Same table as `IcebergPreCreatedFixture`, with a sink that delivers at
least once and stores
+/// no offset within a test, so every message it committed is above the stored
offset. Its
+/// batches hold [`DEFERRED_CHECKPOINT_BATCH_LENGTH`] messages, so that a few
messages already
+/// make a backlog of several batches.
+pub struct IcebergDeferredCheckpointFixture {
+ inner: IcebergPreCreatedFixture,
+}
+
+impl std::ops::Deref for IcebergDeferredCheckpointFixture {
+ type Target = IcebergFixture;
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+impl IcebergOps for IcebergDeferredCheckpointFixture {
+ fn catalog_url(&self) -> &str {
+ self.inner.catalog_url()
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ self.inner.http_client()
+ }
+}
+
+#[async_trait]
+impl TestFixture for IcebergDeferredCheckpointFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ Ok(Self {
+ inner: IcebergPreCreatedFixture::setup().await?,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let mut envs = self.inner.connectors_runtime_envs();
+ envs.insert(
+ ENV_SINK_DELIVERY_GUARANTEE.to_string(),
+ "at_least_once".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_DELIVERY_FLUSH_INTERVAL.to_string(),
+ "1h".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_STREAM.to_string(),
+ seeds::names::STREAM.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_TOPICS.to_string(),
+ format!("[{}]", seeds::names::TOPIC),
+ );
+ envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string());
+ envs.insert(
+ ENV_SINK_STREAMS_0_BATCH_LENGTH.to_string(),
+ DEFERRED_CHECKPOINT_BATCH_LENGTH.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(),
+ "iceberg_sink_connector".to_string(),
+ );
+ envs
+ }
+}
+
/// Same table as `IcebergPreCreatedFixture`, partitioned by
identity(`active`).
pub struct IcebergPartitionedTableFixture {
inner: IcebergPreCreatedFixture,
diff --git a/core/integration/tests/connectors/fixtures/iceberg/mod.rs
b/core/integration/tests/connectors/fixtures/iceberg/mod.rs
index d50c55079..2b9ce93f1 100644
--- a/core/integration/tests/connectors/fixtures/iceberg/mod.rs
+++ b/core/integration/tests/connectors/fixtures/iceberg/mod.rs
@@ -18,7 +18,8 @@
mod container;
pub use container::{
- DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergCatalogRestartFixture,
IcebergEnvAuthFixture,
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, DEFERRED_CHECKPOINT_BATCH_LENGTH,
IcebergAtLeastOnceFixture,
+ IcebergCatalogRestartFixture, IcebergDeferredCheckpointFixture,
IcebergEnvAuthFixture,
IcebergOps, IcebergPartitionedTableFixture, IcebergPreCreatedFixture,
IcebergSkipInvalidRowsFixture,
};
diff --git a/core/integration/tests/connectors/fixtures/mod.rs
b/core/integration/tests/connectors/fixtures/mod.rs
index 440b1bc6e..c0d944952 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -72,7 +72,8 @@ pub use http::{
PARTNER_BEARER_TOKEN, PARTNER_ENDPOINT_ID, PARTNER_INSTANCE,
};
pub use iceberg::{
- DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergCatalogRestartFixture,
IcebergEnvAuthFixture,
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, DEFERRED_CHECKPOINT_BATCH_LENGTH,
IcebergAtLeastOnceFixture,
+ IcebergCatalogRestartFixture, IcebergDeferredCheckpointFixture,
IcebergEnvAuthFixture,
IcebergOps, IcebergPartitionedTableFixture, IcebergPreCreatedFixture,
IcebergSkipInvalidRowsFixture,
};
diff --git a/core/integration/tests/connectors/iceberg/iceberg_sink_delivery.rs
b/core/integration/tests/connectors/iceberg/iceberg_sink_delivery.rs
new file mode 100644
index 000000000..e6a2da0df
--- /dev/null
+++ b/core/integration/tests/connectors/iceberg/iceberg_sink_delivery.rs
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! At-least-once delivery of the Iceberg sink. Every test ends with a
sentinel message: the
+//! runtime delivers a partition in order, so once the snapshot that holds the
sentinel is in
+//! the table, every delivery before it is too, and a duplicate shows in the
row count.
+
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{
+ DEFAULT_NAMESPACE, DEFAULT_TABLE, DEFERRED_CHECKPOINT_BATCH_LENGTH,
IcebergAtLeastOnceFixture,
+ IcebergDeferredCheckpointFixture, IcebergOps,
+};
+use bytes::Bytes;
+use iggy::prelude::{IggyClient, IggyMessage, Partitioning};
+use iggy_common::{Identifier, MessageClient};
+use integration::harness::{ConnectorsRuntimeHandle, seeds};
+use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+const POLL_INTERVAL: Duration = Duration::from_millis(500);
+const DELIVERY_TIMEOUT: Duration = Duration::from_secs(60);
+/// A start after a kill waits for the session of the killed runtime to end
before its
+/// partitions move to the new one.
+const RECOVERY_TIMEOUT: Duration = Duration::from_secs(120);
+
+/// The batch that fails while the catalog is down is handed over again by the
runtime once
+/// the sink's own commit retries run out, and lands once the catalog is back.
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/iceberg/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn
given_at_least_once_when_the_catalog_is_down_longer_than_the_commit_retries_should_commit_every_row_once(
+ harness: &TestHarness,
+ fixture: IcebergAtLeastOnceFixture,
+) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("root client should connect");
+ send(&client, json_messages(1, 3)).await;
+ wait_for_commit_of(&fixture, 2, DELIVERY_TIMEOUT).await;
+
+ fixture
+ .stop_catalog()
+ .await
+ .expect("the catalog should stop");
+ send(&client, json_messages(4, 4)).await;
+ wait_for_redelivery(
+ harness
+ .connectors_runtime()
+ .expect("connector runtime should be available"),
+ )
+ .await;
+ fixture
+ .start_catalog()
+ .await
+ .expect("the catalog should start again");
+ send(&client, json_messages(8, 1)).await;
+
+ wait_for_commit_of(&fixture, 7, DELIVERY_TIMEOUT).await;
+ assert_eq!(
+ committed_rows(&fixture).await,
+ 8,
+ "every row should be committed once"
+ );
+}
+
+/// The sink committed every message, in several batches, but the runtime
stored no offset
+/// before it was killed. The restart hands every message over again, starting
with a batch that
+/// does not carry the newest append, and the table must not hold any of them
twice.
+#[iggy_harness(
+ server(
+ heartbeat.interval = "8s",
+ consumer_group.heartbeat_interval = "1s",
+ connectors_runtime(config_path = "tests/connectors/iceberg/sink.toml")
+ ),
+ seed = seeds::connector_stream
+)]
+async fn
given_at_least_once_when_the_runtime_is_killed_before_storing_offsets_should_hold_every_row_once(
+ harness: &mut TestHarness,
+ fixture: IcebergDeferredCheckpointFixture,
+) {
+ let backlog = 3 * DEFERRED_CHECKPOINT_BATCH_LENGTH as usize;
+ let last = backlog as u64 - 1;
+ let client = harness
+ .root_client()
+ .await
+ .expect("root client should connect");
+ send(&client, json_messages(1, backlog)).await;
+ wait_for_commit_of(&fixture, last, DELIVERY_TIMEOUT).await;
+
+ harness
+ .server_mut()
+ .connectors_runtime_mut()
+ .expect("connector runtime should be available")
+ .kill()
+ .expect("the runtime should be killed");
+ harness
+ .server_mut()
+ .start_dependents()
+ .await
+ .expect("the runtime should start again");
+ send(&client, json_messages(backlog as u128 + 1, 1)).await;
+
+ wait_for_commit_of(&fixture, last + 1, RECOVERY_TIMEOUT).await;
+ assert_eq!(
+ committed_rows(&fixture).await,
+ backlog as u64 + 1,
+ "the messages delivered again should not be appended twice"
+ );
+ let skipped_first_batch = format!(
+ "Skipping {DEFERRED_CHECKPOINT_BATCH_LENGTH} messages from offset: 0
to offset: {}",
+ DEFERRED_CHECKPOINT_BATCH_LENGTH - 1
+ );
+ let (logs, _) = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .collect_logs();
+ assert!(
+ logs.contains(&skipped_first_batch),
+ "the restart should deliver the backlog again and skip it, from its
first batch on"
+ );
+}
+
+async fn send(client: &IggyClient, mut messages: Vec<IggyMessage>) {
+ let stream_id: Identifier = seeds::names::STREAM.try_into().expect("valid
stream name");
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().expect("valid
topic name");
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("the messages should be sent");
+}
+
+fn json_messages(first_id: u128, count: usize) -> Vec<IggyMessage> {
+ create_test_messages(count)
+ .iter()
+ .zip(first_id..)
+ .map(|(row, id)| {
+ let payload = serde_json::to_vec(row).expect("the row should
serialize");
+ IggyMessage::builder()
+ .id(id)
+ .payload(Bytes::from(payload))
+ .build()
+ .expect("the message should build")
+ })
+ .collect()
+}
+
+/// The rows that the table's snapshots appended.
+async fn committed_rows(fixture: &impl IcebergOps) -> u64 {
+ let metadata = fixture
+ .get_table_metadata(DEFAULT_NAMESPACE, DEFAULT_TABLE)
+ .await
+ .expect("the table metadata should load");
+ metadata
+ .metadata
+ .snapshots
+ .unwrap_or_default()
+ .iter()
+ .filter_map(|snapshot|
snapshot.summary.as_ref()?.added_records.as_deref())
+ .map(|records| records.parse::<u64>().expect("added-records is a
count"))
+ .sum()
+}
+
+/// Waits until a snapshot records the message at `offset` as the last one it
holds.
+async fn wait_for_commit_of(fixture: &impl IcebergOps, offset: u64, timeout:
Duration) {
+ let deadline = Instant::now() + timeout;
+ let offset = offset.to_string();
+ loop {
+ let committed = fixture
+ .get_table_metadata(DEFAULT_NAMESPACE, DEFAULT_TABLE)
+ .await
+ .ok()
+ .and_then(|metadata| metadata.metadata.snapshots)
+ .unwrap_or_default()
+ .iter()
+ .any(|snapshot| {
+ snapshot
+ .summary
+ .as_ref()
+ .and_then(|summary| summary.last_offset.as_deref())
+ == Some(offset.as_str())
+ });
+ if committed {
+ return;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "no snapshot holds the message at offset {offset}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
+
+/// Waits for the runtime to hand a failed batch over again.
+async fn wait_for_redelivery(runtime: &ConnectorsRuntimeHandle) {
+ let deadline = Instant::now() + DELIVERY_TIMEOUT;
+ while !runtime
+ .collect_logs()
+ .0
+ .contains("Delivering again from offsets")
+ {
+ assert!(
+ Instant::now() < deadline,
+ "the runtime should deliver the failed batch again"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
diff --git a/core/integration/tests/connectors/iceberg/mod.rs
b/core/integration/tests/connectors/iceberg/mod.rs
index 04fbbc7a5..2fad04f23 100644
--- a/core/integration/tests/connectors/iceberg/mod.rs
+++ b/core/integration/tests/connectors/iceberg/mod.rs
@@ -16,3 +16,4 @@
// under the License.
mod iceberg_sink;
+mod iceberg_sink_delivery;
