Skip to main content
Version: Next

ActiveMQ

ActiveMQ Classic queue source connector

Description​

Reads JMS TextMessage payloads from one ActiveMQ Classic queue over OpenWire. Reuses the existing ActiveMQ module and JSON/text deserializers. This is not a generic JMS or AMQP connector.

Support Those Engines​

Flink
SeaTunnel Zeta

The source requires streaming mode and periodic checkpoints. Spark is not supported by this first slice: its virtual source checkpoints have not been validated as an end-to-end commit boundary for destructive queue acknowledgements.

Key Features​

Options​

NameTypeRequiredDefault
uristringyes-
queue_namestringyes-
usernamestringno-
passwordstringno-
formatenumnoJSON
field_delimiterstringno,
max_in_flight_messagesintno1000
schemaconfigyes-

uri [string]​

One broker endpoint: tcp://host:port or ssl://host:port. Credentials, URI query options, composite transports (including failover:), and URI paths are not supported. Use the separate credential options. For TLS, configure the JVM trust/key stores on every worker and use ssl://; never disable certificate validation. Plain TCP should only be used on a trusted network.

The connector fails the task on a broken connection so the engine restores the reader with a new connection. It does not reconnect in place or reuse acknowledgement handles from an earlier session.

queue_name [string]​

One literal queue name. Wildcards, composite queues, destination prefixes and ?consumer.* destination options are rejected. Topics, durable topic subscriptions and selectors are outside this first slice. Provision the queue and the broker account's permissions before starting the job; broker auto-creation policy is not controlled by the connector.

username / password [string]​

Optional credentials, which must be configured together. When omitted, the client connects without explicit credentials. Use a broker account scoped to the intended queue. Do not embed credentials in uri.

format / field_delimiter​

JSON maps an object or an array of objects using schema. TEXT uses the existing delimited text deserializer, with field_delimiter (default ,). Only JMS TextMessage is accepted; ObjectMessage, BytesMessage, null or empty bodies, and invalid payloads fail the task without acknowledging that message. No Java object payload is deserialized. Deserialization/emission failures omit the original exception details because those details can contain the queue payload; verify the schema and format against the message through an authorized broker client.

max_in_flight_messages [int]​

Positive maximum count of emitted, unacknowledged messages per reader. The source pauses receiving when this limit is reached and resumes after checkpoint completion. The ActiveMQ client queue prefetch is set to the same count; client-side prefetched messages may consume additional memory. This is not a byte limit. Size the limit and producer message sizes for available worker memory. Frequent successful checkpoints are needed for sustained throughput.

schema [config]​

Payload schema; see Schema Feature. Message headers/properties are not exposed as metadata in this slice.

Common Options​

See Source Common Options.

Delivery and Recovery​

  • Delivery is at least once, not exactly once. Each message is individually acknowledged only after a checkpoint containing its emitted rows completes. A later completed checkpoint can include messages from earlier aborted checkpoints.
  • Failure before acknowledgement closes the connection and leaves messages eligible for broker redelivery. Duplicates are possible after recovery, including partially emitted multi-row JSON messages; use an idempotent downstream sink where needed.
  • Checkpoint state contains logical consumer slots, not broker message objects or queue offsets. ActiveMQ remains responsible for retaining and redelivering unacknowledged messages.
  • Each active parallel reader owns a connection, session and competing queue consumer. Normal startup assigns one slot per reader. Recovery preserves existing logical slots and adds slots when parallelism increases; a reader can own multiple restored slots while using one consumer after scale-down. No global ordering is guaranteed with parallel consumers or redelivery.
  • An empty queue is not end-of-input. The source continues waiting until the job is cancelled or fails.
  • Retention depends on broker persistence, producer delivery mode, expiration and dead-letter/redelivery policies. A durable broker and persistent messages are required to survive broker restarts. A permanently invalid message can repeatedly fail the job or be routed to a dead-letter queue by broker policy; this connector never silently skips it.

Example​

env {
  parallelism = 2
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  ActiveMQ {
    uri = "tcp://activemq-host:61616"
    queue_name = "events"
    format = JSON
    max_in_flight_messages = 1000
    schema {
      fields {
        id = bigint
        name = string
      }
    }
  }
}

sink {
  Console {}
}

For delimited messages such as 42|created, set format = TEXT and field_delimiter = "|", with schema fields in the payload's field order. Existing ActiveMQ sink configuration and behavior are unchanged; source options do not apply to the sink.

Changelog​

Change Log
ChangeCommitVersion
[Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source-Next
[Improve][connector][activemq] Remove duplicate dependencies (#8753)https://github.com/apache/seatunnel/commit/da6241aa1c2.3.10
[improve] update activemq connector config option (#8580)https://github.com/apache/seatunnel/commit/629f85b23a2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
Bump org.apache.activemq:activemq-client (#7323)https://github.com/apache/seatunnel/commit/e23e3ac4ed2.3.7
[Feature][Activemq] Added activemq sink (#7251)https://github.com/apache/seatunnel/commit/f0cefbeb4a2.3.7