{"record":{"id":"222f1c58f1214d24","repo":"risingwavelabs/risingwave","slug":"pulsar-error","errorCode":null,"errorMessage":"{pulsar::Error}","messagePattern":"\\{pulsar::Error\\}","errorType":"exception","errorClass":"SinkError::Pulsar","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":68,"sourceCode":"\nconst fn _default_max_retries() -> u32 {\n    3\n}\n\nconst fn _default_retry_backoff() -> Duration {\n    Duration::from_millis(100)\n}\n\nconst fn _default_batch_size() -> u32 {\n    10000\n}\n\nconst fn _default_batch_byte_size() -> usize {\n    1 << 20\n}\n\nfn pulsar_to_sink_err(e: pulsar::Error) -> SinkError {\n    SinkError::Pulsar(anyhow!(e))\n}\n\nasync fn build_pulsar_producer(\n    pulsar: &Pulsar<TokioExecutor>,\n    config: &PulsarConfig,\n) -> Result<Producer<TokioExecutor>> {\n    // Reduce async state machine size (see `clippy::large_futures`).\n    Box::pin(\n        pulsar\n            .producer()\n            .with_options(ProducerOptions {\n                batch_size: Some(config.producer_properties.batch_size),\n                batch_byte_size: Some(config.producer_properties.batch_byte_size),\n                routing_policy: pulsar_producer_routing_policy(\n                    config.producer_properties.routing_mode,\n                ),\n                ..Default::default()\n            })","sourceCodeStart":50,"sourceCodeEnd":86,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L50-L86","documentation":"pulsar_to_sink_err converts any pulsar::Error into a SinkError::Pulsar via anyhow!(e), so any failure from the pulsar-rs client (connection, producer creation, send, subscription) surfaces under this message carrying the underlying Pulsar error's Display text.","triggerScenarios":"Any call on the pulsar client/producer that returns pulsar::Error and is mapped with `?` through pulsar_to_sink_err — client connect, build_pulsar_producer, send/batch send failures.","commonSituations":"Wrong pulsar service URL or unreachable broker; missing auth (JWT/token) for a protected cluster; topic does not exist and auto-creation is disabled; TLS misconfiguration; producer dropped because the broker closed the connection.","solutions":["Read the wrapped pulsar::Error text after '{pulsar::Error}' — it names the root cause (ConnectionRefused, AuthenticationError, NotFound, etc.).","Verify the Pulsar `service.url` is reachable from RisingWave (network/firewall/DNS).","If authentication is required, provide the token/auth plugin fields in the sink WITH options.","Confirm the topic exists or enable topic auto-creation on the broker."],"exampleFix":"// before\nWITH (connector = 'pulsar', service.url = 'pulsar://wrong-host:6650', topic = 't')\n// after\nWITH (connector = 'pulsar', service.url = 'pulsar://pulsar:6650', topic = 'persistent://public/default/t')","handlingStrategy":"try-catch","validationCode":"// preflight connectivity\nnc -zv pulsar-host 6650","typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Pulsar(e) => {\n        // inspect wrapped pulsar::Error; classify connection vs auth vs topic-not-found\n    }\n    _ => {}\n}","preventionTips":["Validate service.url and topic before CREATE SINK.","Provide auth token when the broker requires it.","Enable topic auto-creation or pre-create the topic."],"tags":["rust","pulsar","sink","network"],"backgroundTag":"network-request-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"}