{"record":{"id":"2bcf1a1128c6db18","repo":"vectordotdev/vector","slug":"not-connected","errorCode":null,"errorMessage":"Not Connected","messagePattern":"Not Connected","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sinks/amqp/config.rs","lineNumber":209,"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)?;\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":191,"sourceCodeEnd":227,"githubUrl":"https://github.com/vectordotdev/vector/blob/bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013/src/sinks/amqp/config.rs#L191-L227","documentation":"The AMQP sink healthcheck obtains a channel from the shared channel pool and verifies the underlying connection status. If `channel.status().connected()` is false, it returns a BrokenPipe io::Error labeled \"Not Connected\", failing the healthcheck so Vector reports the sink as unhealthy.","triggerScenarios":"Running the sink healthcheck while the AMQP connection is down or dropped — e.g. broker restarted, network drop, or channels.get() returning a channel whose connection was closed.","commonSituations":"RabbitMQ broker unavailable at startup, credentials/network issues killing the connection between build and healthcheck, or idle connection teardown.","solutions":["Verify the AMQP broker is reachable (host/port/credentials) and retry the healthcheck.","Check broker logs for connection closes (auth failure, vhost missing).","Disable healthcheck (`healthcheck.enabled = false`) or rely on reconnect logic if broker availability is transient.","Ensure connection pooling/reconnect settings keep the channel alive."],"exampleFix":"# before\namqp_url = \"amqp://wrong-host:5672/%2f\"\n\n# after\namqp_url = \"amqp://guest:guest@rabbitmq:5672/%2f\" # reachable broker","handlingStrategy":"retry","validationCode":"// before building the sink, probe the broker\nlet reachable = std::net::TcpStream::connect((host, port)).is_ok();\nif !reachable { return Err(\"AMQP broker unreachable\".into()); }","typeGuard":null,"tryCatchPattern":"match sink.healthcheck().await {\n    Err(e) if e.to_string().contains(\"Not Connected\") => {\n        // wait and re-check broker availability before failing startup\n    }\n    other => other?,\n}","preventionTips":["Verify broker host/port/vhost/credentials before deploying.","Enable reconnect/backoff in AMQP config.","Monitor RabbitMQ connection logs for early disconnects."],"tags":["amqp","healthcheck","connection","rabbitmq"],"backgroundTag":"broken-pipe","analyzedSha":"bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013","analyzedAt":"2026-09-16T02:53:35.741Z","contentChangedAt":"2026-09-16T02:53:35.741Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}