fix(connectors): replace overloaded InvalidRecord with distinct error variants

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),
 }

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,42 +202,65 @@ 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);
 
     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(())
 }