{"record":{"id":"6ec203730a21b3e9","repo":"risingwavelabs/risingwave","slug":"create-iceberg-catalog-for-pk-index-sink","errorCode":null,"errorMessage":"create iceberg catalog for pk-index sink","messagePattern":"create iceberg catalog for pk-index sink","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":374,"sourceCode":"            let f = entry.data_file();\n            if let Some(delete_file) = pending.remove(f.file_path()) {\n                delete_file.set_partition(f.partition().clone());\n                if pending.is_empty() {\n                    return Ok(());\n                }\n            }\n        }\n    }\n    Ok(())\n}\n\nasync fn load_catalog_and_table(\n    iceberg_config: &IcebergConfig,\n) -> Result<(Arc<dyn Catalog>, Table)> {\n    let catalog = iceberg_config\n        .create_catalog()\n        .await\n        .map_err(|e| anyhow!(e).context(\"create iceberg catalog for pk-index sink\"))?;\n    let table = iceberg_config\n        .load_table()\n        .await\n        .map_err(|e| anyhow!(e).context(\"load iceberg table for pk-index sink\"))?;\n    Ok((catalog, table))\n}\n\n/// Read every persisted row for this sink, recovering `prev_committed_epoch` and pending commits.\nasync fn recovery(\n    db: &DatabaseConnection,\n    sink_id: SinkId,\n) -> Result<(Option<u64>, Vec<EpochCommit>)> {\n    fail::fail_point!(\"iceberg_v3_recovery_fail\", |_| Err(anyhow::anyhow!(\n        \"injected: iceberg_v3_recovery_fail\"\n    )));\n    let rows = list_sink_states_ordered_by_epoch(db, sink_id)\n        .await\n        .context(\"list pending sink states for pk-index sink recovery\")?;","sourceCodeStart":356,"sourceCodeEnd":392,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L356-L392","documentation":"When the pk-index sink coordinator initializes, it builds an Iceberg catalog from the sink's `IcebergConfig` via `create_catalog()`. Any failure constructing the catalog (bad catalog type, unreachable REST/Hive/Glue endpoint, bad credentials) is wrapped with this context so the failure is attributable to pk-index sink init.","triggerScenarios":"`init` -> `load_catalog_and_table` -> `iceberg_config.create_catalog()` fails: unknown catalog type in config, network/DNS failure to the catalog service, invalid warehouse URI, missing auth credentials.","commonSituations":"Misconfigured `catalog.type` or `catalog.uri` on the Iceberg sink/connection; catalog service (e.g. REST server or Hive metastore) down or unreachable from the meta node; expired cloud credentials (AWS/Glue).","solutions":["Verify the Iceberg connection/sink config: catalog type, URI, warehouse path, and credentials; test reachability from the meta node (curl the catalog URI).","Check meta node logs for the inner error beneath this context to see the underlying catalog error (auth vs network vs unknown type).","Ensure the catalog backend is running and credentials are valid/rotated; fix then restart the sink."],"exampleFix":"// before\nWITH ('catalog.uri'='http://iceberg-rest:8181/x')\n// after: correct catalog uri & type\nWITH ('catalog.type'='rest', 'catalog.uri'='http://iceberg-rest:8181/catalog')","handlingStrategy":"try-catch","validationCode":"// preflight: verify catalog reachability\n// curl -sSf $CATALOG_URI/v1/config > /dev/null || echo \"catalog unreachable\"","typeGuard":null,"tryCatchPattern":"match load_catalog_and_table(&cfg).await {\n    Ok((catalog, table)) => { /* proceed */ }\n    Err(e) if e.to_string().contains(\"create iceberg catalog\") => {\n        tracing::error!(error = ?e, \"catalog init failed; check catalog.type/uri/credentials\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Validate catalog type/URI/warehouse options when creating the Iceberg connection.","Smoke-test catalog connectivity from the meta node before creating sinks.","Rotate and verify cloud credentials before they expire."],"tags":["iceberg","catalog","configuration","initialization"],"backgroundTag":"module-init-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"}