risingwavelabs/risingwave · error · SinkError::Pulsar

{pulsar::Error from delivery future}

Error message

{pulsar::Error from delivery future}

What it means

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.

Source

Thrown at src/connector/src/sink/pulsar.rs:325

    config: PulsarConfig,
}

struct PulsarPayloadWriter<'w> {
    producer: &'w mut Producer<TokioExecutor>,
    config: &'w PulsarConfig,
    add_future: DeliveryFutureManagerAddFuture<'w, PulsarDeliveryFuture>,
}

mod opaque_type {
    use super::*;
    pub type PulsarDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;

    #[define_opaque(PulsarDeliveryFuture)]
    pub(super) fn may_delivery_future(future: SendFuture) -> PulsarDeliveryFuture {
        future.map(|result| {
            result
                .map(|_| ())
                .map_err(|e: pulsar::Error| SinkError::Pulsar(anyhow!(e)))
        })
    }
}
pub use opaque_type::PulsarDeliveryFuture;
use opaque_type::may_delivery_future;

impl PulsarSinkWriter {
    pub async fn new(
        config: PulsarConfig,
        schema: Schema,
        downstream_pk: Vec<usize>,
        format_desc: &SinkFormatDesc,
        db_name: String,
        sink_from_name: String,
    ) -> Result<Self> {
        let formatter = SinkFormatterImpl::new(
            format_desc,
            schema,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the inner pulsar::Error message to identify whether it is connection, auth, or producer related.
2. Verify service.url, TLS, and auth settings in the sink config are correct and that the broker is reachable.
3. Check broker logs for the topic/producer state at the failure time; recreate or fix the topic if it was deleted.
4. RisingWave retries transient errors downstream; if errors persist, recreate the sink after restoring broker connectivity.

// Pre-check broker reachability before starting the sink
let addr = cfg.service_url.trim_start_matches("pulsar://");
assert!(tokio::net::TcpStream::connect((addr_host(addr), 6650)).await.is_ok(), "broker unreachable");

fn is_pulsar_sink_error(err: &SinkError) -> Option<&anyhow::Error> {
    if let SinkError::Pulsar(e) = err { Some(e) } else { None }
}

match delivery_result {
    Err(SinkError::Pulsar(e)) => {
        warn!("pulsar delivery failed: {e:#}");
        // rely on RisingWave sink retry/backoff, or checkpoint-fail after repeated errors
    }
    _ => {}
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• {pulsar::Error}
• {connection_err (pulsar::Error) after retries exhausted}
• Pulsar error: {0}
• SinkError::SqlServer(anyhow!(err))
• sending bulk write command failed, database: {}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/33c61456a5057d6d. Report an issue: GitHub.