hubcio commented on code in PR #4740:
URL: https://github.com/apache/iggy/pull/4740#discussion_r4468512235
##########
core/integration/tests/connectors/iceberg/iceberg_sink.rs:
##########
@@ -406,3 +410,113 @@ async fn
given_invalid_rows_skip_when_a_batch_holds_a_row_that_does_not_match_sh
.sum();
assert_eq!(written, valid_rows.len() as u64);
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/iceberg/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn
given_a_catalog_restart_when_a_commit_fails_should_retry_until_the_batch_is_committed(
+ harness: &TestHarness,
+ fixture: IcebergCatalogRestartFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let partitioning = Partitioning::partition_id(0);
+
+ let mut first_batch = json_messages(1, 3);
+ client
+ .send_messages(&stream_id, &topic_id, &partitioning, &mut first_batch)
+ .await
+ .expect("Failed to send messages");
+ fixture
+ .wait_for_snapshots(
+ DEFAULT_NAMESPACE,
+ DEFAULT_TABLE,
+ 1,
+ SNAPSHOT_POLL_ATTEMPTS,
+ SNAPSHOT_POLL_INTERVAL_MS,
+ )
+ .await
+ .expect("The first batch should be committed");
+
+ fixture
+ .stop_catalog()
+ .await
+ .expect("Failed to stop the catalog");
+ let mut second_batch = json_messages(4, 4);
+ client
+ .send_messages(&stream_id, &topic_id, &partitioning, &mut second_batch)
+ .await
+ .expect("Failed to send messages");
+ // The runtime stores the offset when it reads a batch, so the sink holds
the
+ // second batch once the offset reaches its last message.
+ wait_for_stored_offset(&client, &stream_id, &topic_id, 6).await;
Review Comment:
warning: the restart test accepts a consumed offset before any failed
catalog commit, so it can pass without exercising retries. wait for the first
failed catalog attempt before restarting the container.
##########
core/connectors/sinks/iceberg_sink/src/router/commit.rs:
##########
@@ -0,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?;
Review Comment:
warning: the outer loop restarts exhausted conflict retries, overriding the
table's retry count and timeout. let `Transaction::commit` own conflict retries
and propagate `CatalogCommitConflicts` after exhaustion.
##########
core/connectors/sinks/iceberg_sink/src/router/commit.rs:
##########
@@ -0,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
Review Comment:
warning: the sink retries HTTP 400/401/403 rejections as `Unexpected`, so
permanent catalog failures consume the full retry budget. preserve the response
status at the catalog boundary and exclude permanent client errors from retries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
