ActiveMQ

ActiveMQ sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Description

Write SeaTunnel rows to an ActiveMQ queue. Each row is serialized as a JSON text message. For consuming queues, see the separate ActiveMQ source.

Key features

• batch
• stream
• exactly-once

Options

name	type	required	default value	description
uri	string	yes	-	ActiveMQ broker URL, such as tcp://localhost:61616.
queue_name	string	yes	-	Queue name to write messages to.
username	string	no	-	Username used to create the ActiveMQ connection. If this option is set, password must also be set.
password	string	no	-	Password used to create the ActiveMQ connection. If this option is set, username must also be set.
client_id	string	no	-	JMS client ID used by the connection factory.
check_for_duplicate	boolean	no	-	Whether the ActiveMQ client checks duplicate messages.
always_session_async	boolean	no	-	Whether the ActiveMQ client always uses a separate thread to dispatch messages for each session.
always_sync_send	boolean	no	-	Whether the ActiveMQ producer always uses synchronous sends.
close_timeout	int	no	-	Timeout in milliseconds before closing the connection is considered failed.
dispatch_async	boolean	no	-	Whether the broker dispatches messages asynchronously.
nested_map_and_list_enabled	boolean	no	-	Whether structured message properties and MapMessage entries can contain nested Map and List objects.
warn_about_unstarted_connection_timeout	int	no	-	Timeout in milliseconds before ActiveMQ warns that a connection was not started correctly. Set a value less than 0 to disable the warning in the ActiveMQ client.
consumer_expiry_check_enabled	boolean	no	-	Whether the ActiveMQ client checks message expiration in each MessageConsumer before dispatching messages.
max_thread_pool_size	int	no	1000	Maximum thread pool size for the ActiveMQ connection factory's internal thread pool used for dispatching messages. Defaults to 1000.
send_timeout	int	no	0	Send timeout in milliseconds for the ActiveMQ connection factory. A value of 0 means no timeout. Defaults to 0.
use_compression	boolean	no	false	Whether message bodies are compressed before being sent to the broker. Defaults to false.
connect_response_timeout	int	no	0	Timeout in milliseconds for waiting for the broker's response during connection establishment. A value of 0 means no timeout. Defaults to 0.
delivery_mode	int	no	2	JMS delivery mode: 1 for NON_PERSISTENT or 2 for PERSISTENT. Defaults to 2 (PERSISTENT).
priority	int	no	4	JMS message priority from 0 (lowest) to 9 (highest). Defaults to 4.
producer_window_size	int	no	0	Flow control window size in bytes for async sends. A value of 0 means no flow control. Defaults to 0.
time_to_live	long	no	0	Message time-to-live in milliseconds. A value of 0 means never expires. Defaults to 0.
use_async_send	boolean	no	false	When true, sends return without waiting for broker confirmation. Defaults to false.

Notes

• uri is the connection entry point. Put the broker host and port in this value, for example tcp://activemq-host:61616.
• username and password are optional, but they must be configured together when the broker requires authentication.
• The connector writes each SeaTunnel row as one JSON text message to queue_name. There is no separate format option for this sink.
• Configure the broker address with uri. host and port are not ActiveMQ sink options.
• Use any SeaTunnel source before this sink. The ActiveMQ connector only controls how the final rows are sent to the queue.
• When use_async_send = true or delivery_mode = 1 (NON_PERSISTENT), the ActiveMQ client sends without waiting for broker confirmation and this sink does not register an ExceptionListener, so a failed send may not be surfaced to the job. Use these options only when acceptable message loss is tolerable.
• The factory-level options (max_thread_pool_size, send_timeout, use_compression, connect_response_timeout, producer_window_size, use_async_send) are only applied when explicitly set. When not set, the value from the matching jms.* parameter in the uri (if present) or the ActiveMQ client default is used.

Example

Write fake data to an ActiveMQ queue:

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        id = int
        name = string
      }
    }
    rows = [
      { kind = INSERT, fields = [1, "Alice"] }
      { kind = INSERT, fields = [2, "Bob"] }
    ]
  }
}

sink {
  ActiveMQ {
    uri = "tcp://localhost:61616"
    username = "admin"
    password = "admin"
    queue_name = "testQueue"
  }
}

In streaming mode, the sink keeps the same broker connection open and writes each row as it arrives. Username/password can also be embedded in the uri, for example tcp://admin:admin@localhost:61616:

env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  FakeSource {
    schema = {
      fields {
        id = int
        name = string
      }
    }
    rows = [
      { kind = INSERT, fields = [1, "Alice"] }
    ]
  }
}

sink {
  ActiveMQ {
    uri = "tcp://admin:admin@localhost:61616"
    queue_name = "testQueue"
  }
}

FAQ

Does ActiveMQ sink support topics as well as queues?

No. The current sink writes only to JMS queues identified by queue_name; topic destinations are not supported by the connector factory. If you need publish/subscribe semantics, use a generic JMS connector or a separate ActiveMQ-targeted bridge that maps the upstream rows onto a topic — but stick to queues when you want to drive this connector directly.

How are username and password validated?

They are optional. When set, both must be present (configuring one without the other fails the job). They take effect at the JMS connection factory level, which means they override any credentials already embedded in uri. If your broker requires an account, prefer the explicit username/password options over embedding them in the URL so they appear in job config logs instead of inside the connection string.

What message format does each row become?

Each SeaTunnel row is serialized as one JSON text message sent to the configured queue_name. There is no format option; the JSON shape is fixed by the sink's serializer, so any consumer that wants a different encoding must decode the JSON body itself first.

Is exactly-once delivery supported?

No. The sink is best-effort with bounded reconnect behavior driven by the underlying JMS client. Enable checkpointing at the job level if at-least-once replay from upstream is acceptable.

Changelog

Change Log

Change	Commit	Version
[Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source	-	Next
[Improve][connector][activemq] Remove duplicate dependencies (#8753)	https://github.com/apache/seatunnel/commit/da6241aa1c	2.3.10
[improve] update activemq connector config option (#8580)	https://github.com/apache/seatunnel/commit/629f85b23a	2.3.10
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
Bump org.apache.activemq:activemq-client (#7323)	https://github.com/apache/seatunnel/commit/e23e3ac4ed	2.3.7
[Feature][Activemq] Added activemq sink (#7251)	https://github.com/apache/seatunnel/commit/f0cefbeb4a	2.3.7