{"record":{"id":"52255e99a9356771","repo":"apache/seatunnel","slug":"rabbitmq-01","errorCode":"RABBITMQ-01","errorMessage":"handle queue consumer shutdown signal failed","messagePattern":"handle queue consumer shutdown signal failed","errorType":"error_code","errorClass":"RabbitmqConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/QueueingConsumer.java","lineNumber":77,"sourceCode":"        super(channel);\n        this.queue = queue;\n        this.splitId = splitId;\n    }\n\n    private void checkShutdown() {\n        if (shutdown != null) {\n            throw Utility.fixStackTrace(shutdown);\n        }\n    }\n\n    @Override\n    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {\n        shutdown = sig;\n        try {\n            queue.put(POISON);\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, e);\n        }\n    }\n\n    @SneakyThrows\n    @Override\n    public void handleCancel(String consumerTag) throws IOException {\n        cancelled = new ConsumerCancelledException();\n        queue.put(POISON);\n    }\n\n    @SneakyThrows\n    @Override\n    public void handleDelivery(\n            String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)\n            throws IOException {\n        checkShutdown();\n        Delivery delivery = new Delivery(envelope, properties, body);\n        DeliveryMessage message = new DeliveryMessage(this.splitId, delivery);","sourceCodeStart":59,"sourceCodeEnd":95,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/QueueingConsumer.java#L59-L95","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Inspect the attached ShutdownSignalException cause for the AMQP reason (connection.reset, access refused, channel error)","Enable RabbitMQ connection automatic recovery or add connector-level retry/restart for transient broker shutdowns","Check network stability and broker heartbeat/timeout settings between workers and broker","Reduce downstream backpressure (faster sink, more parallelism) so the consumer queue does not stay full"],"exampleFix":"// before\n} catch (InterruptedException e) {\n    Thread.currentThread().interrupt();\n    throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, e);\n}\n// after\n} catch (InterruptedException e) {\n    Thread.currentThread().interrupt();\n    LOG.warn(\"interrupted while signaling consumer shutdown; original signal: {}\", sig);\n    throw new RabbitmqConnectorException(HANDLE_SHUTDOWN_SIGNAL_FAILED, sig, e);\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    consumer.start(); // registers handleShutdownSignal\n} catch (RabbitmqConnectorException e) {\n    if (e.getErrorCode() == RabbitmqConnectorErrorCode.HANDLE_SHUTDOWN_SIGNAL_FAILED) {\n        Throwable cause = e.getCause();\n        if (cause instanceof ShutdownSignalException) {\n            LOG.error(\"RabbitMQ shutdown signal: {}\", ((ShutdownSignalException) cause).getReason());\n        }\n        Thread.currentThread().interrupt(); // library restored the flag\n    } else throw e;\n}","preventionTips":["Enable automatic connection recovery in the RabbitMQ connection settings","Monitor broker uptime/restarts and align job scheduling with maintenance windows","Keep consumer queue small (healthy downstream sink) to avoid blocking put","Ensure stable network between workers and broker; tune heartbeat/timeout values"],"tags":["rabbitmq","amqp","consumer","shutdown","interruption"],"backgroundTag":"connection-reset-by-peer","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}