{"record":{"id":"53ddaff818410cac","repo":"risingwavelabs/risingwave","slug":"commit-iceberg-transaction","errorCode":null,"errorMessage":"commit iceberg transaction","messagePattern":"commit iceberg transaction","errorType":"exception","errorClass":"CommitError::Commit","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/commit.rs","lineNumber":782,"sourceCode":"                        );\n                        CommitError::Commit(anyhow!(err).context(\"apply iceberg fast_append\"))\n                    })?;\n\n                    let table = tx.commit(catalog.as_ref()).await.map_err(|err| {\n                        let err: IcebergError = err.into();\n                        tracing::error!(\n                            iceberg_component = \"sink_committer\",\n                            iceberg_operation = \"commit\",\n                            sink_id = %sink_id,\n                            table = %table_name,\n                            epoch,\n                            snapshot_id,\n                            branch = %target_branch,\n                            data_file_count,\n                            error = %err.as_report(),\n                            \"iceberg_sink_commit_transaction_failed\",\n                        );\n                        CommitError::Commit(anyhow!(err).context(\"commit iceberg transaction\"))\n                    })?;\n                    Ok(table)\n                }\n            },\n        )\n        .await\n        .map_err(SinkError::Iceberg)?;\n        self.table = table;\n\n        let snapshot_num = self.table.metadata().snapshots().count();\n        let catalog_name = self.config.common.catalog_name();\n        let table_name = self.table.identifier().to_string();\n        let metrics_labels = [&self.param.sink_name, &catalog_name, &table_name];\n        GLOBAL_SINK_METRICS\n            .iceberg_snapshot_num\n            .with_guarded_label_values(&metrics_labels)\n            .set(snapshot_num as i64);\n","sourceCodeStart":764,"sourceCodeEnd":800,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/commit.rs#L764-L800","documentation":"Wraps an error from `Transaction::commit(catalog)` — the actual catalog commit (table metadata update) of the fast-append transaction in the Iceberg sink. Commit failures typically mean the table metadata was concurrently modified (optimistic concurrency conflict) or the catalog rejected the request.","triggerScenarios":"Another writer committed a snapshot between this sink's table load and commit; catalog service returned an HTTP error (commit-conflict-retry, 409/500); authentication or catalog quota issues during the commit PUT/POST; branch requirement conflicts.","commonSituations":"Multiple RisingWave sink fragments or external engines (Spark, Flink) committing to the same table concurrently; REST catalog behind a load balancer with inconsistent metadata; short catalog token expiring mid-commit.","solutions":["Rely on / increase commit_retry::run_with_retry retries — most conflicts are transient; check commit_retry_num config.","Reduce concurrent committers to the same table/branch, or ensure the catalog supports commit conflict retry (e.g., REST catalog with 409 handling).","Inspect the inner catalog error: fix auth (expired token), network, or catalog capacity issues as indicated.","If conflicts are persistent, serialize commits via a single sink or enable catalog-side retries."],"exampleFix":"// before\nlet table = tx.commit(catalog.as_ref()).await.map_err(|err| {\n    CommitError::Commit(anyhow!(err).context(\"commit iceberg transaction\"))\n})?;\n// after: classify conflicts as retryable\nlet table = tx.commit(catalog.as_ref()).await.map_err(|err| {\n    if is_commit_conflict(&err) { CommitError::Retryable(anyhow!(err).context(\"commit iceberg transaction\")) }\n    else { CommitError::Commit(anyhow!(err).context(\"commit iceberg transaction\")) }\n})?;","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match catalog_commit {\n    Err(e) if e.is_commit_conflict() => retry_with_fresh_table_load(),\n    Err(e) => { log_inner_cause(&e); alert(); }\n    Ok(t) => record_committed_snapshot(t),\n}","preventionTips":["Enable retry (commit_retry_num) and use a catalog that retries on 409 conflicts.","Reduce the number of concurrent committers per table/branch.","Refresh auth tokens before long-running sink epochs."],"tags":["iceberg","catalog","concurrency","commit-conflict"],"backgroundTag":"database-write-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}