ActiveMQ

ActiveMQ sink connector

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.

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"
  }
}

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