You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/05/25 10:41:14 UTC
(iggy) branch master updated: fix(connectors): replace overloaded InvalidRecord with distinct error variants (#3194)
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 8819a7801 fix(connectors): replace overloaded InvalidRecord with distinct error variants (#3194)
8819a7801 is described below
commit 8819a780127a5d4813dd00bf9c7fc06bad67da6f
Author: Atharva Lade <92...@users.noreply.github.com>
AuthorDate: Mon May 25 03:41:10 2026 -0700
fix(connectors): replace overloaded InvalidRecord with distinct error variants (#3194)
---
core/connectors/sdk/src/lib.rs | 21 +++++++
.../sinks/iceberg_sink/src/router/mod.rs | 69 ++++++++++++++++------
2 files changed, 73 insertions(+), 17 deletions(-)
diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs
index 2289aa950..627bc2b3d 100644
--- a/core/connectors/sdk/src/lib.rs
+++ b/core/connectors/sdk/src/lib.rs
@@ -425,4 +425,25 @@ pub enum Error {
/// failures so that circuit breakers are not tripped by bad data.
#[error("Permanent HTTP error: {0}")]
PermanentHttpError(String),
+ /// The source schema could not be mapped to the destination schema.
+ /// Indicates a table definition or configuration problem.
+ #[error("Schema mismatch: {0}")]
+ SchemaMismatch(String),
+ /// An I/O failure while writing data (e.g. Parquet serialization, file
+ /// writer close). Distinct from record-level validation errors. May leave
+ /// orphaned partial files on the object store; cleanup is caller-side.
+ #[error("Write failure: {0}")]
+ WriteFailure(String),
+ /// In-memory transaction preparation failed (e.g. invalid partition spec,
+ /// schema validation). Typically deterministic in the current Iceberg
+ /// version; check the underlying Iceberg error to decide retryability.
+ #[error("Transaction apply error: {0}")]
+ TransactionApplyError(String),
+ /// A catalog commit failed. `Transaction::commit()` consumes the
+ /// transaction, so retrying requires rebuilding it from new data files.
+ /// Retry is not idempotent: callers must verify via the catalog whether
+ /// the original commit was applied before retrying, otherwise data may
+ /// be duplicated.
+ #[error("Catalog commit error: {0}")]
+ CatalogCommitError(String),
}
diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
index 8e60beedf..67a4a555c 100644
--- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs
+++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs
@@ -41,6 +41,17 @@ use std::sync::Arc;
use tracing::{error, warn};
use uuid::Uuid;
+fn format_error_chain(err: &dyn std::error::Error) -> String {
+ let mut chain = err.to_string();
+ let mut source = err.source();
+ while let Some(cause) = source {
+ chain.push_str(": ");
+ chain.push_str(&cause.to_string());
+ source = cause.source();
+ }
+ chain
+}
+
mod arrow_streamer;
pub mod dynamic_router;
pub mod static_router;
@@ -173,12 +184,13 @@ async fn write_data(
let cursor = JsonArrowReader::new(msgs.as_slice());
let reader = ReaderBuilder::new(Arc::new(
schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| {
+ let chain = format_error_chain(&err);
error!(
"Error while mapping records to Iceberg table with uuid: {}. Error {}",
table.metadata().uuid(),
- err
+ chain
);
- Error::InvalidRecord
+ Error::SchemaMismatch(chain)
})?,
))
.build(cursor)
@@ -190,20 +202,41 @@ async fn write_data(
Error::InitError(err.to_string())
})?;
- for batch in reader {
- let batch_data = batch.map_err(|err| {
- error!("Error while getting record batch: {}", err);
- Error::InvalidRecord
- })?;
- writer.write(batch_data).await.map_err(|err| {
- error!("Error while writing record batch: {}", err);
- Error::InvalidRecord
- })?;
+ let write_result: Result<(), Error> = async {
+ for batch in reader {
+ let batch_data = batch.map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!("Error while getting record batch: {}", chain);
+ Error::InvalidRecordValue(chain)
+ })?;
+ writer.write(batch_data).await.map_err(|err| {
+ let chain = format_error_chain(&err);
+ error!("Error while writing record batch: {}", chain);
+ Error::WriteFailure(chain)
+ })?;
+ }
+ Ok(())
+ }
+ .await;
+
+ if let Err(e) = &write_result {
+ error!(
+ "Batch loop failed ({}), closing writer to release resources",
+ e
+ );
+ if let Err(close_err) = writer.close().await {
+ error!("Failed to close writer after batch error: {}", close_err);
+ }
+ return Err(write_result.unwrap_err());
}
let data_files = writer.close().await.map_err(|err| {
- error!("Error while writing data records to Parquet file: {}", err);
- Error::InvalidRecord
+ let chain = format_error_chain(&err);
+ error!(
+ "Error while writing data records to Parquet file: {}",
+ chain
+ );
+ Error::WriteFailure(chain)
})?;
let table_commit = Transaction::new(table);
@@ -211,21 +244,23 @@ async fn write_data(
let action = table_commit.fast_append().add_data_files(data_files);
let tx = action.apply(table_commit).map_err(|err| {
+ let chain = format_error_chain(&err);
error!(
"Failed to apply transaction on table with UUID: {}, Error: {}",
table.metadata().uuid(),
- err
+ chain
);
- Error::InvalidRecord
+ Error::TransactionApplyError(chain)
})?;
let _table = tx.commit(catalog).await.map_err(|err| {
+ let chain = format_error_chain(&err);
error!(
"Failed to commit transaction on table with UUID: {}, Error: {}",
table.metadata().uuid(),
- err
+ chain
);
- Error::InvalidRecord
+ Error::CatalogCommitError(chain)
})?;
Ok(())
}