{"record":{"id":"3a0492678966d55a","repo":"vectordotdev/vector","slug":"brokenpipe","errorCode":"BrokenPipe","errorMessage":"Not Connected","messagePattern":"Not Connected","errorType":"exception","errorClass":"std::io::Error","httpStatus":null,"severity":"error","filePath":"src/sinks/amqp/config.rs","lineNumber":218,"sourceCode":"        _cx: SinkContext,\n    ) -> crate::Result<(VectorSink, Healthcheck)> {\n        let ValidatedAmqpSink {\n            exchange,\n            routing_key,\n        } = validated.clone();\n        let sink = AmqpSink::new(self.clone(), exchange, routing_key).await?;\n        let hc = healthcheck(sink.channels.clone()).boxed();\n        Ok((VectorSink::from_event_streamsink(sink), hc))\n    }\n}\n\npub(super) async fn healthcheck(channels: AmqpSinkChannels) -> crate::Result<()> {\n    trace!(\"Healthcheck started.\");\n\n    let channel = channels.get().await?;\n\n    if !channel.status().connected() {\n        return Err(Box::new(std::io::Error::new(\n            std::io::ErrorKind::BrokenPipe,\n            \"Not Connected\",\n        )));\n    }\n\n    trace!(\"Healthcheck completed.\");\n    Ok(())\n}\n\n#[cfg(test)]\nmod tests {\n    use super::*;\n    use crate::config::ValidatedSink;\n    use crate::config::format::{Format, deserialize};\n    use crate::template::{ConfinementConfig, Template};\n    use vrl::event_path;\n\n    #[test]","sourceCodeStart":200,"sourceCodeEnd":236,"githubUrl":"https://github.com/vectordotdev/vector/blob/99894c8d8885659cc16efc39bd71f9fb3b59e296/src/sinks/amqp/config.rs#L200-L236","documentation":"Runtime error from the AMQP (RabbitMQ) sink's healthcheck in src/sinks/amqp/config.rs. The healthcheck grabs a channel from the sink's channel pool and calls `channel.status().connected()`; if the underlying AMQP connection is no longer connected, it returns `io::Error` (kind `BrokenPipe`, message \"Not Connected\"). It means the broker connection died even though a channel object was handed out.","triggerScenarios":"Vector's startup or reload healthcheck against RabbitMQ when the broker restarted, the TCP connection was reset (idle timeout, firewall, LB), authentication/vhost failed after connect, or the connection limit on the broker closed the session.","commonSituations":"RabbitMQ behind load balancers or with aggressive idle timeouts killing long-lived AMQP connections; broker restarts during deploys; `amqp://` URL pointing at the wrong vhost so the connection is immediately torn down.","solutions":["Verify RabbitMQ is reachable and healthy: check broker logs and `rabbitmq-diagnostics` output","Validate the `amqp_uri` (host, port, vhost, credentials, TLS scheme) — a connection that opens then closes usually means vhost/auth/limits","If a proxy/LB kills idle connections, lower the client heartbeat or enable keepalive so the connection stays active","Re-run/retry the healthcheck once the broker is back; Vector re-runs healthchecks rather than treating one failure as permanent"],"exampleFix":"# before\n[sinks.out]\ntype = \"amqp\"\nendpoint = \"amqp://guest:guest@127.0.0.1:5672/%2Fmissing_vhost\"\n\n# after\n[sinks.out]\ntype = \"amqp\"\nendpoint = \"amqp://guest:guest@127.0.0.1:5672/%2F\"","handlingStrategy":"retry","validationCode":"// Pre-flight: broker TCP reachability before deploying the sink\nuse tokio::net::TcpStream;\nlet reach = TcpStream::connect((host.as_str(), port)).await.is_ok();","typeGuard":null,"tryCatchPattern":"match healthcheck(channels).await {\n    Err(e) if matches!(std::io::Error::other(&e).kind(), std::io::ErrorKind::BrokenPipe) => {\n        // broker connection lost: let Vector's healthcheck retry; investigate broker/vhost/auth\n    }\n    other => other,\n}","preventionTips":["Validate amqp_uri (host, port, vhost, credentials) before deploy","Keep RabbitMQ heartbeat/keepalive on when intermediaries drop idle connections","Watch broker connection logs during Vector starts/reloads to catch immediate disconnects"],"tags":["rust","amqp","rabbitmq","healthcheck","network","vector-sink"],"backgroundTag":"broker-connection-lost","analyzedSha":"99894c8d8885659cc16efc39bd71f9fb3b59e296","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}