{"record":{"id":"ed6cef052677923f","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-sink-coordinator-for-sink-time","errorCode":null,"errorMessage":"iceberg pk-index sink coordinator for sink {} timed out after {}s loading iceberg catalog/table","messagePattern":"iceberg pk-index sink coordinator for sink (.+?) timed out after (.+?)s loading iceberg catalog/table","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":112,"sourceCode":"    retry_num: usize,\n    /// The epoch pre-committed but not yet committed, carried from `pre_commit` to the next `commit`.\n    waiting_commit: Option<EpochCommit>,\n    prev_committed_epoch: Option<u64>,\n}\n\nimpl IcebergPkIndexSinkCoordinator {\n    /// Build a ready-to-serve coordinator: load the iceberg catalog/table, recover any persisted pending\n    /// state, and drain recovered pending epochs to iceberg. Returns only once recovery is complete, so a\n    /// successful return means the sink is ready to accept live pre-commit/commit calls.\n    pub async fn init(\n        sink_id: SinkId,\n        iceberg_config: IcebergConfig,\n        db: DatabaseConnection,\n    ) -> Result<Self> {\n        let (catalog, table) = timeout(INIT_TIMEOUT, load_catalog_and_table(&iceberg_config))\n            .await\n            .map_err(|_| {\n                anyhow!(\n                    \"iceberg pk-index sink coordinator for sink {} timed out after {}s loading iceberg catalog/table\",\n                    sink_id,\n                    INIT_TIMEOUT.as_secs()\n                )\n            })?\n            .with_context(|| format!(\"init iceberg pk-index sink coordinator for sink {}\", sink_id))?;\n\n        let (prev_committed_epoch, recovered) =\n            recovery(&db, sink_id).await.with_context(|| {\n                format!(\n                    \"recover pending state for iceberg pk-index sink {}\",\n                    sink_id\n                )\n            })?;\n\n        let target_branch =\n            commit_branch(iceberg_config.r#type.as_str(), iceberg_config.write_mode);\n","sourceCodeStart":94,"sourceCodeEnd":130,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L94-L130","documentation":"IcebergPkIndexSinkCoordinator::init fails when load_catalog_and_table does not complete within the fixed 60s INIT_TIMEOUT. This means connecting to the Iceberg catalog and loading the target table took too long; the coordinator cannot be built so the sink cannot start serving commits.","triggerScenarios":"Creating/starting an Iceberg pk-index sink: the coordinator's init calls timeout(60s, load_catalog_and_table(&iceberg_config)) and the elapsed timer fires before the catalog/table load returns.","commonSituations":"Wrong or unreachable catalog endpoint (REST/Hive/Glue) in iceberg config, missing or wrong credentials causing auth retries, DNS/network issues in the cluster, very large table metadata (tens of thousands of snapshots/files) making load slow, catalog service outage.","solutions":["Verify iceberg catalog configuration (endpoint URL, warehouse, credentials) and test reachability from the meta node","Fix network/DNS or firewall issues blocking the catalog endpoint","Check catalog service health; retry sink creation after the catalog recovers","Reduce table metadata load (expire old snapshots) if the table is huge and the load legitimately exceeds 60s","Check the wrapped context 'init iceberg pk-index sink coordinator for sink {}' error for the underlying cause"],"exampleFix":"// before (config)\niceberg.endpoint = \"http://internal-rest:8181\" // unreachable -> 60s timeout\n// after\niceberg.endpoint = \"http://iceberg-rest.catalog.svc:8181\" // verified reachable\n// then: curl http://iceberg-rest.catalog.svc:8181/v1/config before creating the sink","handlingStrategy":"retry","validationCode":"// Before creating the sink, verify catalog reachability and config completeness\nasync fn validate_iceberg_catalog(cfg: &IcebergConfig) -> Result<()> {\n    ensure!(!cfg.endpoint.is_empty(), \"iceberg catalog endpoint is empty\");\n    let resp = reqwest::get(format!(\"{}/v1/config\", cfg.endpoint)).await?;\n    ensure!(resp.status().is_success(), \"catalog endpoint not healthy\");\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":"match IcebergPkIndexSinkCoordinator::init(sink_id, cfg.clone(), db.clone()).await {\n    Err(e) if e.to_string().contains(\"timed out after 60s\") => {\n        warn!(\"catalog load timed out; checking endpoint then retrying\");\n        validate_iceberg_catalog(&cfg).await?;\n        retry_with_backoff(|| init(sink_id, cfg.clone(), db.clone())).await\n    }\n    other => other,\n}","preventionTips":["Pre-flight check catalog endpoint reachability and credentials from the meta node","Resolve table metadata bloat: expire snapshots and orphan files regularly","Keep the 60s INIT_TIMEOUT in mind; huge tables may need catalog-side tuning","Monitor catalog service health before sink creation/restart"],"tags":["iceberg","timeout","catalog","initialization"],"backgroundTag":"request-timeout","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"}