{"record":{"id":"cb4564ae4fc128ca","repo":"risingwavelabs/risingwave","slug":"connection-err-pulsar-error-after-retries-exha","errorCode":null,"errorMessage":"{connection_err (pulsar::Error) after retries exhausted}","messagePattern":"\\{connection_err \\(pulsar::Error\\) after retries exhausted\\}","errorType":"exception","errorClass":"SinkError::Pulsar","httpStatus":null,"severity":"critical","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":400,"sourceCode":"                    success_flag = true;\n                    break;\n                }\n                // error upon sending\n                Err(e) => match e {\n                    pulsar::Error::Connection(_)\n                    | pulsar::Error::Producer(_)\n                    | pulsar::Error::Consumer(_) => {\n                        connection_err = Some(e);\n                        tokio::time::sleep(self.config.retry_interval).await;\n                        continue;\n                    }\n                    _ => return Err(SinkError::Pulsar(anyhow!(e))),\n                },\n            }\n        }\n\n        if !success_flag {\n            Err(SinkError::Pulsar(anyhow!(connection_err.unwrap())))\n        } else {\n            Ok(())\n        }\n    }\n\n    async fn write_inner(\n        &mut self,\n        event_key_object: Option<String>,\n        event_object: Option<Vec<u8>>,\n    ) -> Result<()> {\n        let message = Message {\n            partition_key: event_key_object,\n            payload: event_object.unwrap_or_default(),\n            ..Default::default()\n        };\n\n        self.send_message(message).await?;\n        Ok(())","sourceCodeStart":382,"sourceCodeEnd":418,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L382-L418","documentation":"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.","triggerScenarios":"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.","commonSituations":"Pulsar broker/cluster outage or network firewall blocking the service.url port; DNS resolution failure; broker load-shedding producers; persistent TLS handshake failure.","solutions":["Confirm the Pulsar broker is reachable from RisingWave (network, DNS, firewall on the pulsar:// port).","Validate service.url, TLS and auth settings in the sink WITH options.","Increase retry_interval/attempt budget if bursts of broker unavailability are expected (config-dependent).","Check broker logs and metrics at the failure window, then recreate or resume the sink once connectivity is restored."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Health-check the broker before sink activation\nlet addrs: Vec<&str> = cfg.service_url.split(',').collect();\nfor a in addrs {\n    assert!(tcp_probe(a).await.is_ok(), \"pulsar broker {a} unreachable\");\n}","typeGuard":"fn is_connection_class_error(e: &SinkError) -> bool {\n    matches!(e, SinkError::Pulsar(err) if format!(\"{err:#}\") .contains(\"Connection\"))\n}","tryCatchPattern":"match Err(SinkError::Pulsar(e)) = result {\n    // After retries exhausted, treat as outage: back off and retry sink recovery,\n    // or fail the checkpoint so upstream pauses rather than losing data.\n    error!(\"pulsar sink retries exhausted: {e:#}\");\n}","preventionTips":["Ensure broker ports (6650, 8080) are open from RisingWave hosts","Set sensible retry_interval for bursty unavailability","Monitor broker health/load-shedding metrics","Use redundant brokers/advertised addresses"],"tags":["rust","pulsar","sink","network","retry"],"backgroundTag":"connection-refused","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"}