{"record":{"id":"412a988586544509","repo":"risingwavelabs/risingwave","slug":"iceberg-catalog-did-not-reflect-committed-pk-index","errorCode":null,"errorMessage":"iceberg catalog did not reflect committed pk-index snapshot {:?} after {} attempts (last current_snapshot_id={:?})","messagePattern":"iceberg catalog did not reflect committed pk-index snapshot (.+?) after (.+?) attempts \\(last current_snapshot_id=(.+?)\\)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/mod.rs","lineNumber":65,"sourceCode":"pub async fn load_table_at_least(\n    config: &IcebergConfig,\n    expected: Option<i64>,\n) -> SinkResult<Table> {\n    const MAX_ATTEMPTS: usize = 10;\n    const BACKOFF: Duration = Duration::from_millis(500);\n    let mut last = None;\n    for _ in 0..MAX_ATTEMPTS {\n        let table = config.load_table().await?;\n        let Some(expected) = expected else {\n            return Ok(table);\n        };\n        if table.metadata().snapshot_by_id(expected).is_some() {\n            return Ok(table);\n        }\n        last = Some(table.metadata().current_snapshot_id());\n        tokio::time::sleep(BACKOFF).await;\n    }\n    Err(SinkError::Iceberg(anyhow::anyhow!(\n        \"iceberg catalog did not reflect committed pk-index snapshot {:?} after {} attempts (last current_snapshot_id={:?})\",\n        expected,\n        MAX_ATTEMPTS,\n        last,\n    )))\n}\n","sourceCodeStart":47,"sourceCodeEnd":72,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/mod.rs#L47-L72","documentation":"load_table_at_least polls the Iceberg catalog until its loaded table metadata includes the expected committed snapshot id (the pk-index sink's committed snapshot) and gives up after MAX_ATTEMPTS with this error. It exists because a commit by the sink may not be immediately visible to the merger's catalog load (eventual consistency / propagation lag). If the catalog still shows an older current_snapshot_id after all retries, seeding the merger's position-delete staging is aborted.","triggerScenarios":"Raised by load_table_at_least (called from resolve and start_seed) when, after MAX_ATTEMPTS backoff sleeps, table.metadata().snapshot_by_id(expected) is still None and current_snapshot_id remains below/behind the expected snapshot id returned by wait_iceberg_pk_index_sink_epoch.","commonSituations":"Catalog backends with cached/laggy metadata (e.g. REST catalogs with stale cache, or a catalog whose refresh interval exceeds the retry window); the pk-index sink commit failed silently so the expected snapshot never exists; network issues between merger and catalog; clocks/attempts too tight for a heavily loaded catalog.","solutions":["Verify the pk-index sink actually committed the expected snapshot (check sink metrics/logs and the catalog's table history).","Check catalog health and reduce metadata staleness (disable/shorten catalog cache TTL, ensure REST catalog doesn't serve stale metadata).","Inspect network connectivity between the merger node and the catalog service.","Increase MAX_ATTEMPTS or BACKOFF if the catalog is known to be slow under load.","Retry the job — a transient catalog hiccup may resolve on re-seed."],"exampleFix":"// before\nErr(SinkError::Iceberg(anyhow::anyhow!(\n    \"iceberg catalog did not reflect committed pk-index snapshot {:?} after {} attempts ...\"\n)))\n// after\n// No code fix; retry the sink/merger after confirming the catalog is reachable\n// and that the sink's commit succeeded, e.g.:\n// SELECT * FROM \"iceberg_catalog_history\" WHERE table = '...';  -- verify snapshot exists","handlingStrategy":"retry","validationCode":"// Rust (operator-side check before relying on merger seed)\nlet table = catalog.load_table(&ident).await?;\nlet expected = /* committed snapshot id from sink */;\nif table.metadata().snapshot_by_id(expected).is_none() {\n    eprintln!(\"catalog lagging: current={:?}, expected={:?}\",\n        table.metadata().current_snapshot_id(), expected);\n}","typeGuard":"fn catalog_has_snapshot(table: &Table, expected: i64) -> bool {\n    table.metadata().snapshot_by_id(expected).is_some()\n}","tryCatchPattern":"// wrap seeding with retry-and-diagnose\nmatch start_seed(...).await {\n    Err(SinkError::Iceberg(e)) if format!(\"{:e}\", e).contains(\"did not reflect committed pk-index snapshot\") => {\n        warn!(\"catalog lagging or commit missing; verify sink commit then retry\");\n    }\n    other => other?,\n}","preventionTips":["Monitor catalog metadata staleness and refresh latency.","Alert on sink commit epochs not advancing.","Ensure catalog client caches have short TTLs.","Keep BACKOFF/MAX_ATTEMPTS sized for your catalog's worst-case propagation."],"tags":["iceberg","catalog","eventual-consistency","retry-exhausted","streaming-sink"],"backgroundTag":"unexpected-response-shape","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}