risingwave_connector/sink/iceberg/
commit_retry.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Shared iceberg commit-with-retry primitive used by both the V1/V2 sink
16//! committer and the V3 sink coordinator worker. Wraps the standard pattern
17//! of "reload the table, build an action against the freshly-loaded snapshot,
18//! commit, retry on transient errors only".
19
20use std::future::Future;
21use std::sync::Arc;
22use std::time::Duration;
23
24use anyhow::{Result, anyhow, bail};
25use iceberg::table::Table;
26use iceberg::{Catalog, TableIdent};
27use thiserror_ext::AsReport;
28use tokio_retry::RetryIf;
29use tokio_retry::strategy::{ExponentialBackoff, jitter};
30
31#[derive(Clone, Debug)]
32pub struct CommitRetryLogContext {
33    pub iceberg_component: &'static str,
34    pub iceberg_operation: &'static str,
35    pub table: String,
36    pub branch: String,
37    pub sink_id: Option<String>,
38    pub epoch: Option<u64>,
39    pub snapshot_id: Option<i64>,
40}
41
42impl CommitRetryLogContext {
43    pub fn new(
44        iceberg_component: &'static str,
45        iceberg_operation: &'static str,
46        table: impl Into<String>,
47        branch: impl Into<String>,
48    ) -> Self {
49        Self {
50            iceberg_component,
51            iceberg_operation,
52            table: table.into(),
53            branch: branch.into(),
54            sink_id: None,
55            epoch: None,
56            snapshot_id: None,
57        }
58    }
59
60    pub fn with_sink_id(mut self, sink_id: impl ToString) -> Self {
61        self.sink_id = Some(sink_id.to_string());
62        self
63    }
64
65    pub fn with_epoch(mut self, epoch: u64) -> Self {
66        self.epoch = Some(epoch);
67        self
68    }
69
70    pub fn with_snapshot_id(mut self, snapshot_id: i64) -> Self {
71        self.snapshot_id = Some(snapshot_id);
72        self
73    }
74}
75
76/// Distinguishes retriable from non-retriable errors inside [`run_with_retry`].
77pub enum CommitError {
78    /// `reload_table` failed (table not found, schema mismatch, partition
79    /// evolution). Non-retriable — the call site's invariants no longer hold.
80    ReloadTable(anyhow::Error),
81    /// `Transaction::commit` (or its `apply`) failed. Retriable — likely a
82    /// commit conflict or transient network error.
83    Commit(anyhow::Error),
84}
85
86/// Reload the iceberg table from the catalog and assert that its current
87/// `schema_id` and `default_partition_spec_id` still match the values the
88/// caller computed against. Schema or partition evolution mid-commit is
89/// surfaced as a non-retriable error by the call sites.
90pub async fn reload_table(
91    catalog: &dyn Catalog,
92    table_ident: &TableIdent,
93    schema_id: i32,
94    partition_spec_id: i32,
95) -> Result<Table> {
96    let table = catalog
97        .load_table(table_ident)
98        .await
99        .map_err(|e| anyhow!(e).context("reload iceberg table"))?;
100    if table.metadata().current_schema_id() != schema_id {
101        bail!(
102            "iceberg sink: schema evolution not supported; expect schema id {}, got {}",
103            schema_id,
104            table.metadata().current_schema_id(),
105        );
106    }
107    if table.metadata().default_partition_spec_id() != partition_spec_id {
108        bail!(
109            "iceberg sink: partition evolution not supported; expect partition spec id {}, got {}",
110            partition_spec_id,
111            table.metadata().default_partition_spec_id(),
112        );
113    }
114    Ok(table)
115}
116
117/// Run a commit-action against the given iceberg table with retry.
118/// 1. Calls `reload_table` before each commit attempt to get the latest metadata
119/// 2. If `reload_table` fails (table not exists/schema/partition mismatch), stops retrying immediately
120/// 3. If commit fails, retries with backoff up to `retry_num` times.
121///
122/// Strategy: exponential backoff 10ms→60s with jitter, up to `retry_num` retries.
123pub async fn run_with_retry<F, Fut, Out>(
124    catalog: Arc<dyn Catalog>,
125    table_ident: TableIdent,
126    schema_id: i32,
127    partition_spec_id: i32,
128    retry_num: usize,
129    log_context: CommitRetryLogContext,
130    commit_action: F,
131) -> Result<Out>
132where
133    F: Fn(Table) -> Fut + Send + Sync,
134    Fut: Future<Output = Result<Out, CommitError>> + Send,
135{
136    let retry_strategy = ExponentialBackoff::from_millis(10)
137        .max_delay(Duration::from_secs(60))
138        .map(jitter)
139        .take(retry_num);
140
141    RetryIf::spawn(
142        retry_strategy,
143        || {
144            let catalog = catalog.clone();
145            let table_ident = table_ident.clone();
146            let commit_action = &commit_action;
147            async move {
148                let table =
149                    reload_table(catalog.as_ref(), &table_ident, schema_id, partition_spec_id)
150                        .await
151                        .map_err(CommitError::ReloadTable)?;
152                commit_action(table).await
153            }
154        },
155        |err: &CommitError| match err {
156            CommitError::Commit(e) => {
157                tracing::warn!(
158                    iceberg_component = log_context.iceberg_component,
159                    iceberg_operation = log_context.iceberg_operation,
160                    sink_id = log_context.sink_id.as_deref().unwrap_or("unknown"),
161                    epoch = ?log_context.epoch,
162                    snapshot_id = ?log_context.snapshot_id,
163                    table = %log_context.table,
164                    branch = %log_context.branch,
165                    schema_id,
166                    partition_spec_id,
167                    error = %e.as_report(),
168                    "iceberg_commit_retryable_error",
169                );
170                true
171            }
172            CommitError::ReloadTable(e) => {
173                tracing::error!(
174                    iceberg_component = log_context.iceberg_component,
175                    iceberg_operation = log_context.iceberg_operation,
176                    sink_id = log_context.sink_id.as_deref().unwrap_or("unknown"),
177                    epoch = ?log_context.epoch,
178                    snapshot_id = ?log_context.snapshot_id,
179                    table = %log_context.table,
180                    branch = %log_context.branch,
181                    schema_id,
182                    partition_spec_id,
183                    error = %e.as_report(),
184                    "iceberg_commit_reload_table_non_retryable_error",
185                );
186                false
187            }
188        },
189    )
190    .await
191    .map_err(|e| match e {
192        CommitError::ReloadTable(e) | CommitError::Commit(e) => e,
193    })
194}
risingwave_connector/sink/iceberg/commit_retry.rs

risingwave_connector/sink/iceberg/
commit_retry.rs
