apache/seatunnel · error · ActivemqConnectorException

SEND_MESSAGE_FAILED

SEND_MESSAGE_FAILED

Error message

Cannot send AMQ message %s at %s

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-activemq/src/main/java/org/apache/seatunnel/connectors/seatunnel/activemq/client/ActivemqClient.java:126

        if (config.get(NESTED_MAP_AND_LIST_ENABLED) != null) {
            factory.setNestedMapAndListEnabled(config.get(NESTED_MAP_AND_LIST_ENABLED));
        }
        return factory;
    }

    public void write(byte[] msg) {
        try {
            this.connection.start();
            Session session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(config.get(QUEUE_NAME));
            MessageProducer producer = session.createProducer(destination);
            String messageBody = new String(msg, StandardCharsets.UTF_8);
            TextMessage objectMessage = session.createTextMessage(messageBody);
            producer.send(objectMessage);

        } catch (JMSException e) {
            throw new ActivemqConnectorException(
                    ActivemqConnectorErrorCode.SEND_MESSAGE_FAILED,
                    String.format(
                            "Cannot send AMQ message %s at %s",
                            config.get(QUEUE_NAME), config.get(CLIENT_ID)),
                    e);
        }
    }

    public void close() {
        try {
            if (connection != null) {
                connection.close();
            }
        } catch (JMSException e) {
            throw new ActivemqConnectorException(
                    ActivemqConnectorErrorCode.CLOSE_CONNECTION_FAILED,
                    String.format(
                            "Error while closing AMQ connection with  %s", config.get(QUEUE_NAME)));

View on GitHub (pinned to cf67b549a7)

Solutions

1. Check broker availability and queue name/client id values in the config at the addresses reported in the message
2. Inspect the wrapped JMSException cause for the exact delivery failure (e.g. SecurityException, InvalidDestinationException)
3. Retry the job after restoring broker connectivity; enable broker-side logging
4. Validate message payload size against broker limits (maxFrameSize)

Example fix

// before
config.set("queue_name", "My Queue"); // space, may not exist
// after
config.set("queue_name", "myQueue"); // existing destination

// preflight
if (!queueExists(brokerUrl, creds, config.get(QUEUE_NAME))) throw new IllegalStateException("Queue missing");

try { writer.write(row); } catch (ActivemqConnectorException e) { if (e.getErrorCode() == SEND_MESSAGE_FAILED && e.getCause() instanceof JMSException) { /* reconnect client, retry with backoff */ } }

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• CREATE_ACTIVEMQ_CLIENT_FAILED
• CLOSE_CONNECTION_FAILED
• Failed to fetch metadata from Gravitino for metadata: %s
• fail get tableSchema:
• Failed to read AmazonDocumentDB data from database [%s], col

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