{"record":{"id":"d27241e8fa93ad3a","repo":"risingwavelabs/risingwave","slug":"commit-iceberg-pk-index-sink-transaction","errorCode":null,"errorMessage":"commit iceberg pk-index sink transaction","messagePattern":"commit iceberg pk-index sink transaction","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":537,"sourceCode":"                    .iter()\n                    .map(&materialize)\n                    .collect::<Result<Vec<_>, _>>()?;\n\n                let txn = Transaction::new(&table);\n                let action = txn\n                    .overwrite_files()\n                    .set_snapshot_id(snapshot_id)\n                    .set_target_branch(target_branch)\n                    .add_data_files(add_files)\n                    .delete_files(overwrite_files);\n                let txn = action.apply(txn).map_err(|err| {\n                    CommitError::Commit(\n                        anyhow!(err).context(\"apply iceberg pk-index sink overwrite_files action\"),\n                    )\n                })?;\n                let table = txn.commit(catalog.as_ref()).await.map_err(|err| {\n                    CommitError::Commit(\n                        anyhow!(err).context(\"commit iceberg pk-index sink transaction\"),\n                    )\n                })?;\n                Ok(table)\n            }\n        },\n    )\n    .await\n    .map_err(CommitError::Commit)\n}\n\n#[derive(Clone, Serialize, Deserialize)]\nstruct IcebergPkIndexSinkAggResult {\n    schema_id: i32,\n    partition_spec_id: i32,\n    data_files: Vec<SerializedDataFile>,\n    delete_files: Vec<SerializedDataFile>,\n    overwrite_files: Vec<SerializedDataFile>,\n}","sourceCodeStart":519,"sourceCodeEnd":555,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L519-L555","documentation":"After the overwrite action is applied to the transaction, `txn.commit(catalog)` performs the CAS commit against the catalog. Failures (commit conflict with a concurrent writer, catalog HTTP error, timeout) are wrapped with this context as `CommitError::Commit`.","triggerScenarios":"`txn.commit(catalog.as_ref())` fails in `commit_one_epoch`: another engine committed to the table concurrently causing a requirements conflict, the catalog is unreachable, or the commit request times out / returns an error.","commonSituations":"Concurrent compaction or Spark jobs committing to the same Iceberg table; REST catalog rate limiting or downtime; long-running commits exceeding catalog request timeouts; repeated conflicts exhausting `retry_num`.","solutions":["Check the inner error: if it's a requirements/commit conflict, ensure the built-in commit retry (run_with_retry) exhausted its attempts and consider raising retry count.","Pause/avoid external writers (compaction jobs) on the sink-managed table, or schedule them outside sink commit windows.","Verify catalog availability and network from the meta node; increase catalog request timeout if commits are large."],"exampleFix":"null","handlingStrategy":"retry","validationCode":"// preflight catalog reachability\n// curl -sSf -X POST $CATALOG_URI/v1/{prefix}/tables/commit -o /dev/null --max-time 10","typeGuard":null,"tryCatchPattern":"match txn.commit(catalog.as_ref()).await {\n    Ok(table) => Ok(table),\n    Err(e) if is_conflict(&e) => {\n        // retryable: reload latest table and re-run run_with_retry\n        Err(CommitError::Commit(anyhow!(e).context(\"commit iceberg pk-index sink transaction\")))\n    }\n    Err(e) => Err(CommitError::Commit(anyhow!(e).context(\"commit iceberg pk-index sink transaction\"))),\n}","preventionTips":["Don't run concurrent compaction/external writers on the sink-managed table.","Size the commit retry loop for realistic conflict rates.","Monitor catalog health and raise request timeouts for large commits."],"tags":["iceberg","commit-conflict","concurrency","catalog"],"backgroundTag":"http-error-response","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"}