risingwavelabs/risingwave · error · SinkError::Pulsar

{pulsar::Error}

Error message

{pulsar::Error}

What it means

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.

Source

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

const fn _default_max_retries() -> u32 {
    3
}

const fn _default_retry_backoff() -> Duration {
    Duration::from_millis(100)
}

const fn _default_batch_size() -> u32 {
    10000
}

const fn _default_batch_byte_size() -> usize {
    1 << 20
}

fn pulsar_to_sink_err(e: pulsar::Error) -> SinkError {
    SinkError::Pulsar(anyhow!(e))
}

async fn build_pulsar_producer(
    pulsar: &Pulsar<TokioExecutor>,
    config: &PulsarConfig,
) -> Result<Producer<TokioExecutor>> {
    // Reduce async state machine size (see `clippy::large_futures`).
    Box::pin(
        pulsar
            .producer()
            .with_options(ProducerOptions {
                batch_size: Some(config.producer_properties.batch_size),
                batch_byte_size: Some(config.producer_properties.batch_byte_size),
                routing_policy: pulsar_producer_routing_policy(
                    config.producer_properties.routing_mode,
                ),
                ..Default::default()
            })

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the wrapped pulsar::Error text after '{pulsar::Error}' — it names the root cause (ConnectionRefused, AuthenticationError, NotFound, etc.).
2. Verify the Pulsar `service.url` is reachable from RisingWave (network/firewall/DNS).
3. If authentication is required, provide the token/auth plugin fields in the sink WITH options.
4. Confirm the topic exists or enable topic auto-creation on the broker.

Example fix

// before
WITH (connector = 'pulsar', service.url = 'pulsar://wrong-host:6650', topic = 't')
// after
WITH (connector = 'pulsar', service.url = 'pulsar://pulsar:6650', topic = 'persistent://public/default/t')

// preflight connectivity
nc -zv pulsar-host 6650

match err {
    SinkError::Pulsar(e) => {
        // inspect wrapped pulsar::Error; classify connection vs auth vs topic-not-found
    }
    _ => {}
}

When it happens

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

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

Related errors

• {pulsar::Error from delivery future}
• {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/222f1c58f1214d24. Report an issue: GitHub.