apache/seatunnel · error · RabbitmqConnectorException

RABBITMQ-04

RABBITMQ-04

Error message

Cannot send RMQ message to queue %s at host %s

What it means

RabbitmqClient.write throws RabbitmqConnectorException(SEND_MESSAGE_FAILED) when publishing a message to the RabbitMQ exchange/queue fails and retry mode is not enabled. The exception wraps the underlying IOException/TimeoutException from the RabbitMQ java client, naming the queue and host that could not receive the message.

Source

Thrown at seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClient.java:172

     */
    public void write(byte[] msg) {
        try {
            if (StringUtils.isEmpty(config.getRoutingKey())) {
                channel.basicPublish("", config.getQueueName(), null, msg);
            } else {
                // not support set returnListener
                channel.basicPublish(
                        config.getExchange(), config.getRoutingKey(), false, false, null, msg);
            }
        } catch (IOException e) {
            if (config.isLogFailuresOnly()) {
                log.error(
                        "Cannot send RMQ message to queue {} at host {}",
                        config.getQueueName(),
                        config.getHost(),
                        e);
            } else {
                throw new RabbitmqConnectorException(
                        SEND_MESSAGE_FAILED,
                        String.format(
                                "Cannot send RMQ message to queue %s at host %s",
                                config.getQueueName(), config.getHost()),
                        e);
            }
        }
    }

    public void close() {
        Exception t = null;
        try {
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
        } catch (IOException | TimeoutException e) {
            t = e;
        }

View on GitHub (pinned to cf67b549a7)

Solutions

  1. Verify broker connectivity from the engine node (ping/telnet host 5672) and check RabbitMQ server logs for connection/channel errors
  2. Confirm queue, virtualHost, username/password in the config are correct with rabbitmqctl list_queues
  3. If retry is configured, keep it enabled so transient publish failures are retried instead of failing the job; otherwise consider network/heartbeat timeout tuning
  4. Increase RabbitMQ max message/frame size if large payloads trigger channel closure
Defensive patterns

Strategy: retry

Validate before calling

// before starting the job
try (com.rabbitmq.client.Connection c = factory.newConnection()) {
  System.out.println("RabbitMQ reachable: " + c.isOpen());
} catch (Exception e) {
  throw new IllegalStateException("Cannot reach RabbitMQ broker", e);
}

Try / catch

try {
  client.write(row);
} catch (RabbitmqConnectorException e) {
  if (isSendMessageFailed(e)) {
    log.warn("RMQ publish failed, will retry", e);
    // backoff and re-publish or rely on checkpoint restart
  } else {
    throw e;
  }
}

Prevention

When it happens

Trigger: channel.basicPublish(...) in write() throws — broker unreachable, channel closed, publish confirm timeout, or queue unavailable. When retry mode is enabled the client only logs; otherwise it throws.

Common situations: RabbitMQ broker restarted or network partition during a sink job; wrong host/port or vhost credentials; message larger than broker max frame; the RMQ channel was closed earlier by an exception so subsequent publishes fail.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/0f507fdaa4cd2dec. Report an issue: GitHub.