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

risingwave_connector/sink/iceberg/
commit_retry.rs
