vectordotdev/vector · error · std::io::Error

BrokenPipe

BrokenPipe

Error message

Not Connected

What it means

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.

Source

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

        _cx: SinkContext,
    ) -> crate::Result<(VectorSink, Healthcheck)> {
        let ValidatedAmqpSink {
            exchange,
            routing_key,
        } = validated.clone();
        let sink = AmqpSink::new(self.clone(), exchange, routing_key).await?;
        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 99894c8d88)

Solutions

1. Verify RabbitMQ is reachable and healthy: check broker logs and `rabbitmq-diagnostics` output
2. Validate the `amqp_uri` (host, port, vhost, credentials, TLS scheme) — a connection that opens then closes usually means vhost/auth/limits
3. If a proxy/LB kills idle connections, lower the client heartbeat or enable keepalive so the connection stays active
4. Re-run/retry the healthcheck once the broker is back; Vector re-runs healthchecks rather than treating one failure as permanent

Example fix

# before
[sinks.out]
type = "amqp"
endpoint = "amqp://guest:guest@127.0.0.1:5672/%2Fmissing_vhost"

# after
[sinks.out]
type = "amqp"
endpoint = "amqp://guest:guest@127.0.0.1:5672/%2F"

// Pre-flight: broker TCP reachability before deploying the sink
use tokio::net::TcpStream;
let reach = TcpStream::connect((host.as_str(), port)).await.is_ok();

match healthcheck(channels).await {
    Err(e) if matches!(std::io::Error::other(&e).kind(), std::io::ErrorKind::BrokenPipe) => {
        // broker connection lost: let Vector's healthcheck retry; investigate broker/vhost/auth
    }
    other => other,
}

When it happens

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

Common situations: 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.

Related errors

• Pong not received in time
• poll_ready must be called first
• Sink not present
• path and query should never fail to parse
• Serializer does not support JSON

AI-assisted analysis of vectordotdev/vector@99894c8d88 (2026-08-20). Data as JSON: /api/errors/3a0492678966d55a. Report an issue: GitHub.