{"record":{"id":"0a94e1c04a31a189","repo":"risingwavelabs/risingwave","slug":"pulsar-error-0","errorCode":null,"errorMessage":"Pulsar error: {0}","messagePattern":"Pulsar error: (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1184,"sourceCode":"        anyhow::Error,\n    ),\n    #[error(\"LanceDB error: {0}\")]\n    LanceDb(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"ElasticSearch/OpenSearch error: {0}\")]\n    ElasticSearchOpenSearch(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"Starrocks error: {0}\")]\n    Starrocks(String),\n    #[error(\"File error: {0}\")]\n    File(String),\n    #[error(\"Pulsar error: {0}\")]\n    Pulsar(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(transparent)]\n    Internal(\n        #[from]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"BigQuery error: {0}\")]\n    BigQuery(\n        #[source]\n        #[backtrace]\n        anyhow::Error,\n    ),\n    #[error(\"DynamoDB error: {0}\")]","sourceCodeStart":1166,"sourceCodeEnd":1202,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1166-L1202","documentation":"A SinkError variant wrapping an anyhow::Error from the pulsar crate. Raised when the Pulsar sink fails to produce messages: producer creation failure, topic not found, send/ack timeouts, or broker connection loss.","triggerScenarios":"Connecting to the Pulsar broker fails or drops, the topic/service URL is wrong, authentication (JWT/token) fails, or produce calls time out without acknowledgment.","commonSituations":"Wrong service URL or port (6650 vs 8080), expired JWT token, topic namespace/tenant misconfigured, broker overload or unacked-message limits hit.","solutions":["Verify the Pulsar service URL and that the broker is reachable from RisingWave.","Check authentication options (token/JWT) are valid and not expired.","Confirm the topic (tenant/namespace/topic) exists and the producer has permissions.","Inspect the wrapped anyhow message; if transient broker issues, retry/restart the sink."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"pulsar-admin topics list public/default # and test connectivity: nc -zv broker 6650","typeGuard":null,"tryCatchPattern":"match err { SinkError::Pulsar(e) => { log::warn!(\"pulsar: {e:#}\"); retry_with_backoff() /* transient broker errors */ } _ => return Err(err) }","preventionTips":["Check broker URL, port, and auth token expiry before creating the sink","Ensure the topic exists and the role has produce permissions","Keep token credentials refreshable, not hard-coded"],"tags":["sink","pulsar","messaging","network"],"backgroundTag":"connection-refused","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"}