apache/seatunnel · error · RabbitmqConnectorException

RABBITMQ-01

RABBITMQ-01

Error message

handle queue consumer shutdown signal failed

What it means

QueueingConsumer.handleShutdownSignal is the RabbitMQ consumer callback invoked by the client library when the channel/connection shuts down. It signals the internal queue with a POISON element; if the thread is interrupted while doing the blocking queue.put, it re-interrupts the thread and throws RabbitmqConnectorException with HANDLE_SHUTDOWN_SIGNAL_FAILED. The original ShutdownSignalException is attached as the cause.

Source

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

        super(channel);
        this.queue = queue;
        this.splitId = splitId;
    }

    private void checkShutdown() {
        if (shutdown != null) {
            throw Utility.fixStackTrace(shutdown);
        }
    }

    @Override
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
        shutdown = sig;
        try {
            queue.put(POISON);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, e);
        }
    }

    @SneakyThrows
    @Override
    public void handleCancel(String consumerTag) throws IOException {
        cancelled = new ConsumerCancelledException();
        queue.put(POISON);
    }

    @SneakyThrows
    @Override
    public void handleDelivery(
            String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
            throws IOException {
        checkShutdown();
        Delivery delivery = new Delivery(envelope, properties, body);
        DeliveryMessage message = new DeliveryMessage(this.splitId, delivery);

View on GitHub (pinned to cf67b549a7)

Solutions

1. Inspect the attached ShutdownSignalException cause for the AMQP reason (connection.reset, access refused, channel error)
2. Enable RabbitMQ connection automatic recovery or add connector-level retry/restart for transient broker shutdowns
3. Check network stability and broker heartbeat/timeout settings between workers and broker
4. Reduce downstream backpressure (faster sink, more parallelism) so the consumer queue does not stay full

Example fix

// before
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, e);
}
// after
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    LOG.warn("interrupted while signaling consumer shutdown; original signal: {}", sig);
    throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, sig, e);
}

try {
    consumer.start(); // registers handleShutdownSignal
} catch (RabbitmqConnectorException e) {
    if (e.getErrorCode() == RabbitmqConnectorErrorCode.HANDLE_SHUTDOWN_SIGNAL_FAILED) {
        Throwable cause = e.getCause();
        if (cause instanceof ShutdownSignalException) {
            LOG.error("RabbitMQ shutdown signal: {}", ((ShutdownSignalException) cause).getReason());
        }
        Thread.currentThread().interrupt(); // library restored the flag
    } else throw e;
}

When it happens

Trigger: RabbitMQ channel/connection shutdown occurs (broker restart, network drop, forced connection close) and, while the POISON delivery is being placed on the bounded queue, the consumer thread is interrupted (queue full and put blocks until interrupted).

Common situations: Broker restart or failover during a running job; network partition between SeaTunnel worker and RabbitMQ; consumer queue backlogged (slow downstream sink) so queue.put blocks, then job cancellation interrupts the thread.

Related errors

• Enumerator thread is interrupted.
• Failed to close Pulsar consumer.
• Interrupted while draining Python source stdout after proces
• Interrupted while draining python source process output
• RABBITMQ-02

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