risingwavelabs/risingwave · critical · SinkError::Pulsar

{connection_err (pulsar::Error) after retries exhausted}

Error message

{connection_err (pulsar::Error) after retries exhausted}

What it means

After the retry loop in send_message exhausts its retry budget (success_flag false), the last recorded connection error (a pulsar::Error captured across Connection/Producer/Consumer failures) is returned as SinkError::Pulsar. This means the sink could not reach or maintain a healthy producer against the Pulsar broker within the allowed retries.

Source

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

                    success_flag = true;
                    break;
                }
                // error upon sending
                Err(e) => match e {
                    pulsar::Error::Connection(_)
                    | pulsar::Error::Producer(_)
                    | pulsar::Error::Consumer(_) => {
                        connection_err = Some(e);
                        tokio::time::sleep(self.config.retry_interval).await;
                        continue;
                    }
                    _ => return Err(SinkError::Pulsar(anyhow!(e))),
                },
            }
        }

        if !success_flag {
            Err(SinkError::Pulsar(anyhow!(connection_err.unwrap())))
        } else {
            Ok(())
        }
    }

    async fn write_inner(
        &mut self,
        event_key_object: Option<String>,
        event_object: Option<Vec<u8>>,
    ) -> Result<()> {
        let message = Message {
            partition_key: event_key_object,
            payload: event_object.unwrap_or_default(),
            ..Default::default()
        };

        self.send_message(message).await?;
        Ok(())

View on GitHub (pinned to 6469eb736d)

Solutions

1. Confirm the Pulsar broker is reachable from RisingWave (network, DNS, firewall on the pulsar:// port).
2. Validate service.url, TLS and auth settings in the sink WITH options.
3. Increase retry_interval/attempt budget if bursts of broker unavailability are expected (config-dependent).
4. Check broker logs and metrics at the failure window, then recreate or resume the sink once connectivity is restored.

// Health-check the broker before sink activation
let addrs: Vec<&str> = cfg.service_url.split(',').collect();
for a in addrs {
    assert!(tcp_probe(a).await.is_ok(), "pulsar broker {a} unreachable");
}

fn is_connection_class_error(e: &SinkError) -> bool {
    matches!(e, SinkError::Pulsar(err) if format!("{err:#}") .contains("Connection"))
}

match Err(SinkError::Pulsar(e)) = result {
    // After retries exhausted, treat as outage: back off and retry sink recovery,
    // or fail the checkpoint so upstream pauses rather than losing data.
    error!("pulsar sink retries exhausted: {e:#}");
}

When it happens

Trigger: send_message loops retrying Connection/Producer/Consumer errors with config.retry_interval sleeps until the loop bound is reached, then unwraps connection_err and returns it — typical when the broker is down, unreachable, or repeatedly resetting connections.

Common situations: Pulsar broker/cluster outage or network firewall blocking the service.url port; DNS resolution failure; broker load-shedding producers; persistent TLS handshake failure.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

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