{"record":{"id":"8869fa7f7a0f8c5f","repo":"apache/seatunnel","slug":"send-message-failed","errorCode":"SEND_MESSAGE_FAILED","errorMessage":"Cannot send AMQ message %s at %s","messagePattern":"Cannot send AMQ message (.+?) at (.+?)","errorType":"error_code","errorClass":"ActivemqConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java","lineNumber":126,"sourceCode":"\n        if (config.get(NESTED_MAP_AND_LIST_ENABLED) != null) {\n            factory.setNestedMapAndListEnabled(config.get(NESTED_MAP_AND_LIST_ENABLED));\n        }\n        return factory;\n    }\n\n    public void write(byte[] msg) {\n        try {\n            this.connection.start();\n            Session session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);\n            Destination destination = session.createQueue(config.get(QUEUE_NAME));\n            MessageProducer producer = session.createProducer(destination);\n            String messageBody = new String(msg, StandardCharsets.UTF_8);\n            TextMessage objectMessage = session.createTextMessage(messageBody);\n            producer.send(objectMessage);\n\n        } catch (JMSException e) {\n            throw new ActivemqConnectorException(\n                    ActivemqConnectorErrorCode.SEND_MESSAGE_FAILED,\n                    String.format(\n                            \"Cannot send AMQ message %s at %s\",\n                            config.get(QUEUE_NAME), config.get(CLIENT_ID)),\n                    e);\n        }\n    }\n\n    public void close() {\n        try {\n            if (connection != null) {\n                connection.close();\n            }\n        } catch (JMSException e) {\n            throw new ActivemqConnectorException(\n                    ActivemqConnectorErrorCode.CLOSE_CONNECTION_FAILED,\n                    String.format(\n                            \"Error while closing AMQ connection with  %s\", config.get(QUEUE_NAME)));","sourceCodeStart":108,"sourceCodeEnd":144,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java#L108-L144","documentation":"ActivemqClient.write converts each SeaTunnelRow's bytes into a UTF-8 TextMessage and sends it via a JMS MessageProducer; if producer.send throws a JMSException it is wrapped as SEND_MESSAGE_FAILED with the queue name and client id. It indicates the message could not be delivered to the AMQ destination.","triggerScenarios":"A JMSException occurs during session/producer creation or producer.send(objectMessage) in write() — e.g. the connection dropped mid-write, the session is closed, the destination is invalid, or the broker rejected the message.","commonSituations":"Broker restart or network blip during a running job; queue deleted while job runs; message size/limit violations; connection/session closed by a prior failure.","solutions":["Check broker availability and queue name/client id values in the config at the addresses reported in the message","Inspect the wrapped JMSException cause for the exact delivery failure (e.g. SecurityException, InvalidDestinationException)","Retry the job after restoring broker connectivity; enable broker-side logging","Validate message payload size against broker limits (maxFrameSize)"],"exampleFix":"// before\nconfig.set(\"queue_name\", \"My Queue\"); // space, may not exist\n// after\nconfig.set(\"queue_name\", \"myQueue\"); // existing destination","handlingStrategy":"retry","validationCode":"// preflight\nif (!queueExists(brokerUrl, creds, config.get(QUEUE_NAME))) throw new IllegalStateException(\"Queue missing\");","typeGuard":null,"tryCatchPattern":"try { writer.write(row); } catch (ActivemqConnectorException e) { if (e.getErrorCode() == SEND_MESSAGE_FAILED && e.getCause() instanceof JMSException) { /* reconnect client, retry with backoff */ } }","preventionTips":["Use failover:// transport so transient broker/network blips are retried by the JMS client","Keep messages within broker maxFrameSize limits","Ensure the destination exists before the job starts","Monitor broker logs for connection drops during job runtime"],"tags":["activemq","jms","send","network"],"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-14T05:17:10.506Z"}