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. This is a sink-only connector; SeaTunnel does not provide an ActiveMQ source connector.

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.

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.

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
[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