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 risingwave_common::util::retry::exponential_backoff;
28use thiserror_ext::AsReport;
29use tokio_retry::RetryIf;
30use tokio_retry::strategy::jitter;
31
32#[derive(Clone, Debug)]
33pub struct CommitRetryLogContext {
34    pub iceberg_component: &'static str,
35    pub iceberg_operation: &'static str,
36    pub table: String,
37    pub branch: String,
38    pub sink_id: Option<String>,
39    pub epoch: Option<u64>,
40    pub snapshot_id: Option<i64>,
41}
42
43impl CommitRetryLogContext {
44    pub fn new(
45        iceberg_component: &'static str,
46        iceberg_operation: &'static str,
47        table: impl Into<String>,
48        branch: impl Into<String>,
49    ) -> Self {
50        Self {
51            iceberg_component,
52            iceberg_operation,
53            table: table.into(),
54            branch: branch.into(),
55            sink_id: None,
56            epoch: None,
57            snapshot_id: None,
58        }
59    }
60
61    pub fn with_sink_id(mut self, sink_id: impl ToString) -> Self {
62        self.sink_id = Some(sink_id.to_string());
63        self
64    }
65
66    pub fn with_epoch(mut self, epoch: u64) -> Self {
67        self.epoch = Some(epoch);
68        self
69    }
70
71    pub fn with_snapshot_id(mut self, snapshot_id: i64) -> Self {
72        self.snapshot_id = Some(snapshot_id);
73        self
74    }
75}
76
77/// Distinguishes retriable from non-retriable errors inside [`run_with_retry`].
78pub enum CommitError {
79    /// `reload_table` failed (table not found, schema mismatch, partition
80    /// evolution). Non-retriable — the call site's invariants no longer hold.
81    ReloadTable(anyhow::Error),
82    /// `Transaction::commit` (or its `apply`) failed. Retriable — likely a
83    /// commit conflict or transient network error.
84    Commit(anyhow::Error),
85}
86
87/// Reload the iceberg table from the catalog and assert that its current
88/// `schema_id` and `default_partition_spec_id` still match the values the
89/// caller computed against. Schema or partition evolution mid-commit is
90/// surfaced as a non-retriable error by the call sites.
91pub async fn reload_table(
92    catalog: &dyn Catalog,
93    table_ident: &TableIdent,
94    schema_id: i32,
95    partition_spec_id: i32,
96) -> Result<Table> {
97    let table = catalog
98        .load_table(table_ident)
99        .await
100        .map_err(|e| anyhow!(e).context("reload iceberg table"))?;
101    if table.metadata().current_schema_id() != schema_id {
102        bail!(
103            "iceberg sink: schema evolution not supported; expect schema id {}, got {}",
104            schema_id,
105            table.metadata().current_schema_id(),
106        );
107    }
108    if table.metadata().default_partition_spec_id() != partition_spec_id {
109        bail!(
110            "iceberg sink: partition evolution not supported; expect partition spec id {}, got {}",
111            partition_spec_id,
112            table.metadata().default_partition_spec_id(),
113        );
114    }
115    Ok(table)
116}
117
118/// Run a commit-action against the given iceberg table with retry.
119/// 1. Calls `reload_table` before each commit attempt to get the latest metadata
120/// 2. If `reload_table` fails (table not exists/schema/partition mismatch), stops retrying immediately
121/// 3. If commit fails, retries with backoff up to `retry_num` times.
122///
123/// Strategy: exponential backoff 10ms→60s with jitter, up to `retry_num` retries.
124pub async fn run_with_retry<F, Fut, Out>(
125    catalog: Arc<dyn Catalog>,
126    table_ident: TableIdent,
127    schema_id: i32,
128    partition_spec_id: i32,
129    retry_num: usize,
130    log_context: CommitRetryLogContext,
131    commit_action: F,
132) -> Result<Out>
133where
134    F: Fn(Table) -> Fut + Send + Sync,
135    Fut: Future<Output = Result<Out, CommitError>> + Send,
136{
137    let retry_strategy =
138        exponential_backoff(Duration::from_millis(10), 10, Duration::from_secs(60))
139            .map(jitter)
140            .take(retry_num);
141
142    RetryIf::spawn(
143        retry_strategy,
144        || {
145            let catalog = catalog.clone();
146            let table_ident = table_ident.clone();
147            let commit_action = &commit_action;
148            async move {
149                let table =
150                    reload_table(catalog.as_ref(), &table_ident, schema_id, partition_spec_id)
151                        .await
152                        .map_err(CommitError::ReloadTable)?;
153                commit_action(table).await
154            }
155        },
156        |err: &CommitError| match err {
157            CommitError::Commit(e) => {
158                tracing::warn!(
159                    iceberg_component = log_context.iceberg_component,
160                    iceberg_operation = log_context.iceberg_operation,
161                    sink_id = log_context.sink_id.as_deref().unwrap_or("unknown"),
162                    epoch = ?log_context.epoch,
163                    snapshot_id = ?log_context.snapshot_id,
164                    table = %log_context.table,
165                    branch = %log_context.branch,
166                    schema_id,
167                    partition_spec_id,
168                    error = %e.as_report(),
169                    "iceberg_commit_retryable_error",
170                );
171                true
172            }
173            CommitError::ReloadTable(e) => {
174                tracing::error!(
175                    iceberg_component = log_context.iceberg_component,
176                    iceberg_operation = log_context.iceberg_operation,
177                    sink_id = log_context.sink_id.as_deref().unwrap_or("unknown"),
178                    epoch = ?log_context.epoch,
179                    snapshot_id = ?log_context.snapshot_id,
180                    table = %log_context.table,
181                    branch = %log_context.branch,
182                    schema_id,
183                    partition_spec_id,
184                    error = %e.as_report(),
185                    "iceberg_commit_reload_table_non_retryable_error",
186                );
187                false
188            }
189        },
190    )
191    .await
192    .map_err(|e| match e {
193        CommitError::ReloadTable(e) | CommitError::Commit(e) => e,
194    })
195}
risingwave_connector/sink/iceberg/commit_retry.rs

risingwave_connector/sink/iceberg/
commit_retry.rs
