{"record":{"id":"6d24c4f2371683ca","repo":"apache/seatunnel","slug":"rabbitmq-02-6d24c4","errorCode":"RABBITMQ-02","errorMessage":"create rabbitmq client failed","messagePattern":"create rabbitmq client 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/source/RabbitmqSourceReader.java","lineNumber":223,"sourceCode":"        for (RabbitmqSplit split : splits) {\n            log.info(\"Received split for queue: {}\", split.splitId());\n            try {\n                if (activeConsumers.containsKey(split.splitId())) {\n                    log.warn(\"Consumer for queue {} already exists, skipping\", split.splitId());\n                    continue;\n                }\n\n                // Create a new consumer that feeds messages into the shared internal 'queue'\n                DefaultConsumer consumer =\n                        rabbitMQClient.getQueueingConsumer(queue, split.splitId());\n                rabbitMQClient.setupQueue(split.splitId());\n                channel.basicConsume(split.splitId(), autoAck, consumer);\n                activeConsumers.put(split.splitId(), consumer);\n                sourceSplits.add(split);\n\n                log.info(\"Started consuming from queue: {}\", split.splitId());\n            } catch (IOException e) {\n                throw new RabbitmqConnectorException(\n                        org.apache.seatunnel.connectors.seatunnel.rabbitmq.exception\n                                .RabbitmqConnectorErrorCode.CREATE_RABBITMQ_CLIENT_FAILED,\n                        e);\n            }\n        }\n    }\n\n    @Override\n    public List<RabbitmqSplit> snapshotState(long checkpointId) throws Exception {\n        List<Long> deliveryTags =\n                pendingDeliveryTagsToCommit.computeIfAbsent(checkpointId, id -> new ArrayList<>());\n        Set<String> correlationIds =\n                pendingCorrelationIdsToCommit.computeIfAbsent(checkpointId, id -> new HashSet<>());\n        deliveryTags.addAll(deliveryTagsProcessedForCurrentSnapshot);\n        correlationIds.addAll(correlationIdsProcessedButNotAcknowledged);\n        deliveryTagsProcessedForCurrentSnapshot.clear();\n\n        return new ArrayList<>(sourceSplits);","sourceCodeStart":205,"sourceCodeEnd":241,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceReader.java#L205-L241","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before: queue may not exist, basicConsume throws IOException\nchannel.basicConsume(split.splitId(), autoAck, consumer);\n// after: ensure the queue exists before consuming\nchannel.queueDeclare(split.splitId(), true, false, false, null);\nchannel.basicConsume(split.splitId(), autoAck, consumer);","handlingStrategy":"try-catch","validationCode":"// before job start\ntry (Socket s = new Socket(host, port)) { /* broker reachable */ }\n// and ensure the queue exists:\nchannel.queueDeclarePassive(queueName);","typeGuard":null,"tryCatchPattern":"try { reader.addSplits(splits); } catch (RabbitmqConnectorException e) {\n  if (e.getCause() instanceof IOException) { /* reconnect broker / recreate channel, then retry */ }\n  throw e;\n}","preventionTips":["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"],"tags":["rabbitmq","network","io","connector"],"backgroundTag":"network-request-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}