What causes "create rabbitmq client failed" in apache/seatunnel?

Verify the RabbitMQ broker is reachable and the queue named split.splitId() exists before starting the job

apache/seatunnel · error · RabbitmqConnectorException

RABBITMQ-02

RABBITMQ-02

Error message

create rabbitmq client failed

What it means

Thrown by RabbitmqSourceReader.addSplits when opening a RabbitMQ consumer via channel.basicConsume fails with an IOException. It wraps the broker I/O error in a RabbitmqConnectorException with error code CREATE_RABBITMQ_CLIENT_FAILED, indicating the reader could not start consuming from the split's queue.

Source

Thrown at seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceReader.java:223

        for (RabbitmqSplit split : splits) {
            log.info("Received split for queue: {}", split.splitId());
            try {
                if (activeConsumers.containsKey(split.splitId())) {
                    log.warn("Consumer for queue {} already exists, skipping", split.splitId());
                    continue;
                }

                // Create a new consumer that feeds messages into the shared internal 'queue'
                DefaultConsumer consumer =
                        rabbitMQClient.getQueueingConsumer(queue, split.splitId());
                rabbitMQClient.setupQueue(split.splitId());
                channel.basicConsume(split.splitId(), autoAck, consumer);
                activeConsumers.put(split.splitId(), consumer);
                sourceSplits.add(split);

                log.info("Started consuming from queue: {}", split.splitId());
            } catch (IOException e) {
                throw new RabbitmqConnectorException(
                        org.apache.seatunnel.connectors.seatunnel.rabbitmq.exception
                                .RabbitmqConnectorErrorCode.CREATE_RABBITMQ_CLIENT_FAILED,
                        e);
            }
        }
    }

    @Override
    public List<RabbitmqSplit> snapshotState(long checkpointId) throws Exception {
        List<Long> deliveryTags =
                pendingDeliveryTagsToCommit.computeIfAbsent(checkpointId, id -> new ArrayList<>());
        Set<String> correlationIds =
                pendingCorrelationIdsToCommit.computeIfAbsent(checkpointId, id -> new HashSet<>());
        deliveryTags.addAll(deliveryTagsProcessedForCurrentSnapshot);
        correlationIds.addAll(correlationIdsProcessedButNotAcknowledged);
        deliveryTagsProcessedForCurrentSnapshot.clear();

        return new ArrayList<>(sourceSplits);

View on GitHub (pinned to cf67b549a7)

Solutions

Verify the RabbitMQ broker is reachable and the queue named split.splitId() exists before starting the job
Check host/port/vhost/username/password config; confirm the user has consume permission on the queue
Inspect the wrapped cause (e.getCause()) in logs for the actual AMQP error (404 queue-not-found, 403 access-refused, connection reset)
Increase connection/channel recovery settings or heartbeat timeout in the connection factory to survive transient network blips
Enable broker-side logs to confirm whether the broker closed the channel and why

Example fix

// before: queue may not exist, basicConsume throws IOException
channel.basicConsume(split.splitId(), autoAck, consumer);
// after: ensure the queue exists before consuming
channel.queueDeclare(split.splitId(), true, false, false, null);
channel.basicConsume(split.splitId(), autoAck, consumer);

Defensive patterns

Strategy: try-catch

Validate before calling

// before job start
try (Socket s = new Socket(host, port)) { /* broker reachable */ }
// and ensure the queue exists:
channel.queueDeclarePassive(queueName);

Try / catch

try { reader.addSplits(splits); } catch (RabbitmqConnectorException e) {
  if (e.getCause() instanceof IOException) { /* reconnect broker / recreate channel, then retry */ }
  throw e;
}

Prevention

Pre-declare all queues (durable) before submitting the job
Verify broker host/port/vhost/credentials with redis-like smoke test (e.g. rabbitmqctl / management API)
Monitor broker availability and keep connections alive with heartbeats
Check user permissions for consume on the target queues

When it happens

Trigger: Calling addSplits (during source initialization or recovery) when the channel is not open, the queue does not exist, broker connection is down, or AMQP protocol/frame errors occur during basicConsume.

Common situations: RabbitMQ broker restarted or unreachable mid-job; queue deleted before the job starts (passive consume on missing queue); wrong vhost/credentials limiting access; network partition between worker and broker; channel closed by broker due to heartbeat timeout.

Related errors

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