This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch iceberg_catalog_retry in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5a1114386f70e21be9a9b8a2542b1170c309f685 Merge: 0381e1f16 bf6459690 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Nov 4 11:29:59 2026 +0100 merge master Co-Authored-By: Claude Opus 5.5 <[email protected]> .github/workflows/_build_python_wheels.yml | 10 +- .github/workflows/_common.yml | 4 +- .github/workflows/_publish_rust_crates.yml | 8 +- .github/workflows/_test_examples.yml | 4 +- .github/workflows/coverage-baseline.yml | 6 +- .github/workflows/post-merge.yml | 2 +- .github/workflows/publish.yml | 2 +- Cargo.lock | 75 ++-- Cargo.toml | 11 +- README.md | 2 +- bdd/python/uv.lock | 2 +- core/ai/mcp/Cargo.toml | 2 +- core/bench/Cargo.toml | 2 +- core/bench/dashboard/frontend/Cargo.toml | 2 +- core/bench/dashboard/server/Cargo.toml | 2 +- core/bench/report/Cargo.toml | 2 +- core/binary_protocol/Cargo.toml | 2 +- core/certs/iggy_cert.pem | 20 +- core/cli/Cargo.toml | 2 +- core/common/Cargo.toml | 2 +- core/common/src/http/messages/send_messages.rs | 202 ++++++++++- core/common/src/types/message/iggy_message.rs | 53 +++ core/common/src/types/message/mod.rs | 5 +- core/common/src/types/message/user_headers.rs | 391 ++++++++++++++++++++- core/common/src/wire_conversions.rs | 18 +- core/connectors/runtime/Cargo.toml | 2 +- .../example_config/connectors/iceberg_sink.toml | 5 +- core/connectors/sdk/Cargo.toml | 3 +- core/connectors/sdk/src/log.rs | 27 +- core/connectors/sdk/src/sink.rs | 8 +- core/connectors/sdk/src/source.rs | 8 +- core/connectors/sdk/tests/plugin_logging.rs | 100 ++++++ core/connectors/sinks/bigquery_sink/Cargo.toml | 2 +- core/connectors/sinks/clickhouse_sink/Cargo.toml | 2 +- core/connectors/sinks/delta_sink/Cargo.toml | 6 +- core/connectors/sinks/delta_sink/src/coercions.rs | 27 +- core/connectors/sinks/delta_sink/src/lib.rs | 4 +- core/connectors/sinks/delta_sink/src/sink.rs | 75 +++- core/connectors/sinks/doris_sink/Cargo.toml | 2 +- .../connectors/sinks/elasticsearch_sink/Cargo.toml | 2 +- core/connectors/sinks/http_sink/Cargo.toml | 2 +- core/connectors/sinks/iceberg_sink/Cargo.toml | 3 +- core/connectors/sinks/iceberg_sink/README.md | 29 +- core/connectors/sinks/iceberg_sink/config.toml | 2 +- core/connectors/sinks/iceberg_sink/src/lib.rs | 9 +- core/connectors/sinks/iceberg_sink/src/props.rs | 31 +- .../sinks/iceberg_sink/src/router/commit.rs | 3 + .../iceberg_sink/src/router/dynamic_router.rs | 22 +- core/connectors/sinks/iceberg_sink/src/sink.rs | 7 +- core/connectors/sinks/influxdb_sink/Cargo.toml | 2 +- core/connectors/sinks/meilisearch_sink/Cargo.toml | 2 +- core/connectors/sinks/mongodb_sink/Cargo.toml | 2 +- core/connectors/sinks/postgres_sink/Cargo.toml | 2 +- core/connectors/sinks/quickwit_sink/Cargo.toml | 2 +- core/connectors/sinks/rabbitmq_sink/Cargo.toml | 2 +- core/connectors/sinks/redshift_sink/Cargo.toml | 2 +- core/connectors/sinks/s3_sink/Cargo.toml | 2 +- core/connectors/sinks/stdout_sink/Cargo.toml | 2 +- core/connectors/sinks/surrealdb_sink/Cargo.toml | 2 +- .../sources/elasticsearch_source/Cargo.toml | 2 +- core/connectors/sources/http_source/Cargo.toml | 2 +- core/connectors/sources/influxdb_source/Cargo.toml | 2 +- core/connectors/sources/mqtt_source/Cargo.toml | 2 +- core/connectors/sources/postgres_source/Cargo.toml | 5 +- core/connectors/sources/postgres_source/src/lib.rs | 6 +- core/connectors/sources/random_source/Cargo.toml | 2 +- .../tests/server/consumer_offset_recovery.rs | 142 ++++++++ core/integration/tests/server/http_vsr.rs | 83 +++++ core/integration/tests/server/mod.rs | 1 + core/partitions/src/iggy_partition.rs | 18 + core/partitions/src/lib.rs | 9 +- core/partitions/src/partition_storage.rs | 180 ++++++---- core/sdk/Cargo.toml | 2 +- core/sdk/src/links/mux.rs | 14 +- core/sdk/src/prelude.rs | 34 +- core/server/Cargo.toml | 2 +- core/server/config.toml | 1 + core/server/server.http | 17 +- core/server/src/http/extractor.rs | 40 ++- core/server/src/http/handlers.rs | 12 +- core/server/src/partition_helpers.rs | 19 +- examples/python/uv.lock | 2 +- foreign/cpp/Cargo.toml | 2 +- foreign/cpp/MODULE.bazel | 2 +- foreign/cpp/src/type_conversion.rs | 18 +- foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +- foreign/go/contracts/version.go | 2 +- foreign/java/gradle.properties | 2 +- foreign/java/gradle/libs.versions.toml | 2 +- foreign/node/package-lock.json | 4 +- foreign/node/package.json | 2 +- foreign/php/Cargo.toml | 2 +- foreign/python/Cargo.toml | 4 +- foreign/python/pyproject.toml | 2 +- foreign/python/src/user_headers.rs | 7 +- foreign/python/tests/test_message_operations.py | 4 +- foreign/python/uv.lock | 2 +- foreign/swift/Sources/Iggy/Iggy.swift | 2 +- foreign/swift/Tests/IggyTests/Fixtures/golden.json | 8 +- gateways/kafka/Cargo.toml | 2 +- scripts/prepare-release.sh | 15 +- web/package-lock.json | 4 +- web/package.json | 2 +- 103 files changed, 1539 insertions(+), 380 deletions(-) diff --cc core/connectors/sinks/iceberg_sink/README.md index c10bb6413,4deda0a19..1538e0e20 --- a/core/connectors/sinks/iceberg_sink/README.md +++ b/core/connectors/sinks/iceberg_sink/README.md @@@ -120,29 -114,36 +120,54 @@@ over into multiple files, including fo Each destination table has its own append transaction. Fan-out is not atomic across tables: a write or commit error stops work on the remaining tables, while earlier commits remain. -The Iceberg library refreshes metadata before committing and retries eligible commit errors -according to the table's retry properties. The plugin does not replay failed batches. The -runtime records a plugin error and continues polling with consumer auto-commit, so messages -can be lost after failed or skipped writes. Replaying committed rows can create duplicates. +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. + +## Commit Retries + +The sink writes the data files of a batch once, then retries the catalog commit when the catalog +cannot be reached or answers with an unexpected status, such as a 5xx. It makes up to +`max_retries` attempts in total and waits about `retry_delay` before the first retry, doubling the +wait up to `max_retry_delay`. A commit that the catalog refuses for good, such as one to a table +that no longer exists, fails without a retry. + +Each commit tags its snapshot with an `iggy.commit-id` summary property. Before a retry, the sink +reloads the table and stops if a snapshot with that id is already there, so a commit whose response +was lost is not appended twice. A commit request that the catalog applies only after the sink has +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. + ## Commit Rate and Table Maintenance + + Each batch that the runtime hands to the sink becomes one append commit per destination table, + which adds a snapshot and a manifest to the table. Commits slow down as the table metadata grows, + so fewer, larger batches keep both the commit latency and the metadata small. Set the batch size on + the stream: + + ```toml + [[streams]] + stream = "events" + topics = ["users"] + schema = "json" + batch_length = 1000 # most messages in one batch + min_count = 1000 # wait until this many messages are ready, + max_wait = "5s" # or this long, whichever comes first + ``` + + At 2,000 messages per second, `batch_length = 100` creates about 20 snapshots per second, and the + 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. + + 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. + ## Source Compatibility Use `schema = "json"` on the stream. Each JSON object represents a table row; nested objects diff --cc core/connectors/sinks/iceberg_sink/src/lib.rs index 0d9c86403,f428c210d..64f6a2162 --- a/core/connectors/sinks/iceberg_sink/src/lib.rs +++ b/core/connectors/sinks/iceberg_sink/src/lib.rs @@@ -16,14 -16,10 +16,15 @@@ // under the License. use crate::router::Router; +use humantime::Duration as HumanDuration; +use iggy_connector_sdk::retry::RetryPolicy; use iggy_connector_sdk::{Error, sink_connector}; + use secrecy::SecretString; use serde::{Deserialize, Serialize}; +use std::str::FromStr; +use std::time::Duration; use strum::Display as StrumDisplay; +use tracing::warn; mod catalog; mod props; diff --cc core/connectors/sinks/iceberg_sink/src/router/commit.rs index 66b4b015f,000000000..c85c2f6d3 mode 100644,000000..100644 --- a/core/connectors/sinks/iceberg_sink/src/router/commit.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/commit.rs @@@ -1,396 -1,0 +1,399 @@@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::router::format_error_chain; +use iceberg::spec::DataFile; +use iceberg::table::Table; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::{Catalog, ErrorKind, TableIdent}; +use iggy_connector_sdk::Error; +use iggy_connector_sdk::retry::{RetryPolicy, retry_async}; +use std::collections::HashMap; +use tracing::{error, info}; +use uuid::Uuid; + +/// Snapshot summary property naming the append, so that a retry can tell +/// 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. +pub(crate) async fn commit_data_files( + table: &Table, + catalog: &dyn Catalog, + data_files: Vec<DataFile>, + retry: RetryPolicy, +) -> Result<(), Error> { + let commit_id = Uuid::new_v4().to_string(); + 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(), + )])) + .add_data_files(data_files); + let transaction = action.apply(transaction).map_err(|err| { + let chain = format_error_chain(&err); + error!( + "Failed to apply transaction on table with UUID: {}, Error: {}", + table.metadata().uuid(), + chain + ); + Error::TransactionApplyError(chain) + })?; + + // Data files are kept on commit failure: the catalog may have applied the + // commit before the error surfaced, and deleting files referenced by a + // committed snapshot would corrupt the table. + let context = format!("Commit to Iceberg table {}", table.identifier()); + let identifier = table.identifier(); + let transaction = &transaction; + let commit_id = commit_id.as_str(); + let mut attempted = false; + retry_async(retry, &context, is_transient, move || { + let retrying = std::mem::replace(&mut attempted, true); + commit_once(catalog, identifier, transaction, commit_id, retrying) + }) + .await + .map_err(|failure| { + let chain = format_error_chain(&failure.error); + error!( + "Failed to commit transaction on table with UUID: {} after {} attempts, Error: {}", + table.metadata().uuid(), + failure.attempts, + chain + ); + Error::CatalogCommitError(chain) + }) +} + +/// A commit that could not reach the catalog, got an unexpected answer from +/// it, or kept conflicting with other writers can pass on another attempt. +fn is_transient(error: &iceberg::Error) -> bool { + matches!( + error.kind(), + ErrorKind::Unexpected | ErrorKind::CatalogCommitConflicts + ) +} + +async fn commit_once( + catalog: &dyn Catalog, + identifier: &TableIdent, + transaction: &Transaction, + commit_id: &str, + retrying: bool, +) -> Result<(), iceberg::Error> { + if retrying { + let current = catalog.load_table(identifier).await?; + if holds_commit(¤t, commit_id) { + info!( + "Iceberg table {identifier} already holds commit {commit_id} from an earlier attempt" + ); + return Ok(()); + } + } + transaction.clone().commit(catalog).await?; + Ok(()) +} + +fn holds_commit(table: &Table, commit_id: &str) -> bool { + table.metadata().snapshots().any(|snapshot| { + snapshot + .summary() + .additional_properties + .get(COMMIT_ID_PROPERTY) + .is_some_and(|id| id == commit_id) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + 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::{ + CatalogBuilder, MemoryCatalog, Namespace, NamespaceIdent, TableCommit, TableCreation, + }; + use std::collections::VecDeque; + use std::sync::Mutex; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; + + const FAST_RETRY: RetryPolicy = RetryPolicy { + max_attempts: 3, + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + }; + + /// What `FlakyCatalog` does with the next `update_table` call. + #[derive(Debug, Clone, Copy)] + enum CommitFailure { + /// The request never reaches the catalog. + Unreachable, + /// The catalog applies the commit, and the response is lost. + ResponseLost, + /// The catalog refuses the commit for good. + Refused, + } + + /// A memory catalog whose `update_table` fails as scripted, then succeeds. + #[derive(Debug)] + struct FlakyCatalog { + inner: MemoryCatalog, + failures: Mutex<VecDeque<CommitFailure>>, + update_calls: AtomicU32, + } + + impl FlakyCatalog { + async fn with_table(failures: &[CommitFailure]) -> (Self, Table) { + let inner = 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()); + inner + .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"); + let table = inner + .create_table( + &namespace, + TableCreation::builder() + .name("events".to_owned()) + .schema(schema) + .build(), + ) + .await + .expect("Failed to create table"); + let catalog = Self { + inner, + failures: Mutex::new(failures.iter().copied().collect()), + update_calls: AtomicU32::new(0), + }; + (catalog, table) + } + + fn update_calls(&self) -> u32 { + self.update_calls.load(Ordering::Relaxed) + } + + async fn snapshot_count(&self, table: &Table) -> usize { + self.inner + .load_table(table.identifier()) + .await + .expect("Failed to load table") + .metadata() + .snapshots() + .count() + } + } + + #[async_trait] + impl Catalog for FlakyCatalog { + async fn list_namespaces( + &self, + parent: Option<&NamespaceIdent>, + ) -> iceberg::Result<Vec<NamespaceIdent>> { + self.inner.list_namespaces(parent).await + } + + async fn create_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap<String, String>, + ) -> iceberg::Result<Namespace> { + self.inner.create_namespace(namespace, properties).await + } + + async fn get_namespace(&self, namespace: &NamespaceIdent) -> iceberg::Result<Namespace> { + self.inner.get_namespace(namespace).await + } + + async fn namespace_exists(&self, namespace: &NamespaceIdent) -> iceberg::Result<bool> { + self.inner.namespace_exists(namespace).await + } + + async fn update_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap<String, String>, + ) -> iceberg::Result<()> { + self.inner.update_namespace(namespace, properties).await + } + + async fn drop_namespace(&self, namespace: &NamespaceIdent) -> iceberg::Result<()> { + self.inner.drop_namespace(namespace).await + } + + async fn list_tables( + &self, + namespace: &NamespaceIdent, + ) -> iceberg::Result<Vec<TableIdent>> { + self.inner.list_tables(namespace).await + } + + async fn create_table( + &self, + namespace: &NamespaceIdent, + creation: TableCreation, + ) -> iceberg::Result<Table> { + self.inner.create_table(namespace, creation).await + } + + async fn load_table(&self, table: &TableIdent) -> iceberg::Result<Table> { + self.inner.load_table(table).await + } + + async fn drop_table(&self, table: &TableIdent) -> iceberg::Result<()> { + self.inner.drop_table(table).await + } + + async fn purge_table(&self, table: &TableIdent) -> iceberg::Result<()> { + self.inner.purge_table(table).await + } + + async fn table_exists(&self, table: &TableIdent) -> iceberg::Result<bool> { + self.inner.table_exists(table).await + } + + async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> iceberg::Result<()> { + self.inner.rename_table(src, dest).await + } + + async fn register_table( + &self, + table: &TableIdent, + metadata_location: String, + ) -> iceberg::Result<Table> { + self.inner.register_table(table, metadata_location).await + } + + async fn update_table(&self, commit: TableCommit) -> iceberg::Result<Table> { + self.update_calls.fetch_add(1, Ordering::Relaxed); + let failure = self + .failures + .lock() + .expect("Failures lock poisoned") + .pop_front(); + match failure { + None => self.inner.update_table(commit).await, + Some(CommitFailure::Unreachable) => Err(iceberg::Error::new( + ErrorKind::Unexpected, + "Connection refused", + )), + Some(CommitFailure::ResponseLost) => { + self.inner.update_table(commit).await?; + Err(iceberg::Error::new( + ErrorKind::Unexpected, + "Gateway timeout, the commit state is unknown", + )) + } + Some(CommitFailure::Refused) => Err(iceberg::Error::new( + ErrorKind::TableNotFound, + "Tried to update a table that does not exist", + )), + } + } + } + + fn data_file(table: &Table) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(format!( + "{}/data/{}.parquet", + table.metadata().location(), + Uuid::new_v4() + )) + .file_format(DataFileFormat::Parquet) + .record_count(3) + .file_size_in_bytes(100) + .build() + .expect("Failed to build data file") + } + + #[tokio::test] + async fn given_an_unreachable_catalog_should_retry_until_the_commit_lands() { + let (catalog, table) = + 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"); + + assert_eq!(catalog.update_calls(), 3); + assert_eq!(catalog.snapshot_count(&table).await, 1); + } + + #[tokio::test] + 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"); + + assert_eq!(catalog.update_calls(), 1); + assert_eq!(catalog.snapshot_count(&table).await, 1); + } + + #[tokio::test] + 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}"); + assert_eq!(catalog.update_calls(), 1); + assert_eq!(catalog.snapshot_count(&table).await, 0); + } + + #[tokio::test] + 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"); + + assert!(matches!(error, Error::CatalogCommitError(_)), "{error}"); + assert_eq!(catalog.update_calls(), FAST_RETRY.max_attempts); + assert_eq!(catalog.snapshot_count(&table).await, 0); + } +}
