{"record":{"id":"33c61456a5057d6d","repo":"risingwavelabs/risingwave","slug":"pulsar-error-from-delivery-future","errorCode":null,"errorMessage":"{pulsar::Error from delivery future}","messagePattern":"\\{pulsar::Error from delivery future\\}","errorType":"exception","errorClass":"SinkError::Pulsar","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":325,"sourceCode":"    config: PulsarConfig,\n}\n\nstruct PulsarPayloadWriter<'w> {\n    producer: &'w mut Producer<TokioExecutor>,\n    config: &'w PulsarConfig,\n    add_future: DeliveryFutureManagerAddFuture<'w, PulsarDeliveryFuture>,\n}\n\nmod opaque_type {\n    use super::*;\n    pub type PulsarDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;\n\n    #[define_opaque(PulsarDeliveryFuture)]\n    pub(super) fn may_delivery_future(future: SendFuture) -> PulsarDeliveryFuture {\n        future.map(|result| {\n            result\n                .map(|_| ())\n                .map_err(|e: pulsar::Error| SinkError::Pulsar(anyhow!(e)))\n        })\n    }\n}\npub use opaque_type::PulsarDeliveryFuture;\nuse opaque_type::may_delivery_future;\n\nimpl PulsarSinkWriter {\n    pub async fn new(\n        config: PulsarConfig,\n        schema: Schema,\n        downstream_pk: Vec<usize>,\n        format_desc: &SinkFormatDesc,\n        db_name: String,\n        sink_from_name: String,\n    ) -> Result<Self> {\n        let formatter = SinkFormatterImpl::new(\n            format_desc,\n            schema,","sourceCodeStart":307,"sourceCodeEnd":343,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L307-L343","documentation":"The Pulsar delivery future resolves to a Result; may_delivery_future maps any pulsar::Error from the message delivery (broker/producer send acknowledgment) into SinkError::Pulsar. This surfaces transport, producer, or broker-side failures that occur asynchronously when the send future completes. It is the standard mapping point for delivery errors inside send_message's buffered send pipeline.","triggerScenarios":"A send future returned by pulsar producer.send_async (wrapped via may_delivery_future) completes with Err(e: pulsar::Error) — e.g. connection dropped to the broker, authentication failure, producer closed/fenced, topic not found, or message serialization at the client level.","commonSituations":"Pulsar broker restart or network partition during sink writes; wrong service.url or TLS/auth credentials; Pulsar producer being fenced by another producer with the same name; topic deleted while sink is running.","solutions":["Inspect the inner pulsar::Error message to identify whether it is connection, auth, or producer related.","Verify service.url, TLS, and auth settings in the sink config are correct and that the broker is reachable.","Check broker logs for the topic/producer state at the failure time; recreate or fix the topic if it was deleted.","RisingWave retries transient errors downstream; if errors persist, recreate the sink after restoring broker connectivity."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Pre-check broker reachability before starting the sink\nlet addr = cfg.service_url.trim_start_matches(\"pulsar://\");\nassert!(tokio::net::TcpStream::connect((addr_host(addr), 6650)).await.is_ok(), \"broker unreachable\");","typeGuard":"fn is_pulsar_sink_error(err: &SinkError) -> Option<&anyhow::Error> {\n    if let SinkError::Pulsar(e) = err { Some(e) } else { None }\n}","tryCatchPattern":"match delivery_result {\n    Err(SinkError::Pulsar(e)) => {\n        warn!(\"pulsar delivery failed: {e:#}\");\n        // rely on RisingWave sink retry/backoff, or checkpoint-fail after repeated errors\n    }\n    _ => {}\n}","preventionTips":["Monitor broker connectivity and producer stats","Set correct auth/TLS in service.url before creating the sink","Keep broker and client versions compatible","Alert on repeated Pulsar sink errors in logs"],"tags":["rust","pulsar","network","sink","delivery"],"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"}