vectordotdev/vector · error

Not Connected

Error message

Not Connected

What it means

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.

Solutions

1. Verify the AMQP broker is reachable (host/port/credentials) and retry the healthcheck.
2. Check broker logs for connection closes (auth failure, vhost missing).
3. Disable healthcheck (`healthcheck.enabled = false`) or rely on reconnect logic if broker availability is transient.
4. Ensure connection pooling/reconnect settings keep the channel alive.

Example fix

# before
amqp_url = "amqp://wrong-host:5672/%2f"

# after
amqp_url = "amqp://guest:guest@rabbitmq:5672/%2f" # reachable broker

// before building the sink, probe the broker
let reachable = std::net::TcpStream::connect((host, port)).is_ok();
if !reachable { return Err("AMQP broker unreachable".into()); }

match sink.healthcheck().await {
    Err(e) if e.to_string().contains("Not Connected") => {
        // wait and re-check broker availability before failing startup
    }
    other => other?,
}

When it happens

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

Common situations: RabbitMQ broker unavailable at startup, credentials/network issues killing the connection between build and healthcheck, or idle connection teardown.

Related errors

• BrokenPipe
• Building request never fails.
• HTTP status codes are valid u16 values
• Sink not present

AI-assisted analysis of vectordotdev/vector@bdb87aeaa4 (2026-09-16). Data as JSON: /api/errors/2bcf1a1128c6db18. Report an issue: GitHub.

Appendix: source

Thrown at src/sinks/amqp/config.rs:209

        _cx: SinkContext,
    ) -> crate::Result<(VectorSink, Healthcheck)> {
        let ValidatedAmqpSink {
            exchange,
            routing_key,
        } = validated.clone();
        let sink = AmqpSink::new(self.clone(), exchange, routing_key)?;
        let hc = healthcheck(sink.channels.clone()).boxed();
        Ok((VectorSink::from_event_streamsink(sink), hc))
    }
}

pub(super) async fn healthcheck(channels: AmqpSinkChannels) -> crate::Result<()> {
    trace!("Healthcheck started.");

    let channel = channels.get().await?;

    if !channel.status().connected() {
        return Err(Box::new(std::io::Error::new(
            std::io::ErrorKind::BrokenPipe,
            "Not Connected",
        )));
    }

    trace!("Healthcheck completed.");
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::ValidatedSink;
    use crate::config::format::{Format, deserialize};
    use crate::template::{ConfinementConfig, Template};
    use vrl::event_path;

    #[test]

View on GitHub (pinned to bdb87aeaa4)