ActiveMQ Source Node

The activemqsource node consumes messages from an Apache ActiveMQ broker over JMS. It supports both broker families — ActiveMQ Classic and ActiveMQ Artemis — and reads from either a queue or a durable topic subscription. Messages are consumed inside a transacted JMS session, so acknowledgement is tied to the pipeline's commit strategy.

Key Features

• Both broker families: CLASSIC selects the ActiveMQ Classic client, ARTEMIS the Artemis client
• Queues and durable topics: consume a queue, or create a named durable subscription on a topic so messages published while the pipeline is down are retained
• Transacted acknowledgement: the JMS session is transacted, so messages are acknowledged only when the pipeline commits — an uncommitted batch is redelivered after a restart
• Configurable commit strategy: acknowledge per record, per batch/interval, or never
• Batched receive: block briefly for the first message, then drain whatever else is immediately available up to a cap
• Text and bytes messages: TextMessage and BytesMessage bodies are both read
• Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration

Field	Type	Required	Default	Description
brokerUrl	String	Yes	—	Broker URL, e.g. tcp://localhost:61616
brokerType	String	Yes	—	Client to use: CLASSIC or ARTEMIS
destination	String	Yes	—	Queue or topic name to consume from
destinationType	String	No	QUEUE	QUEUE or TOPIC
encodingType	String	Yes	—	Encoding of the message body. See Encoding Types
credentialId	String	No	—	ID of a username/password credential in jobContext.otherProperties. Omit to connect anonymously
clientId	String	Conditional	—	JMS client id. Required when destinationType is TOPIC
subscriptionName	String	Conditional	—	Durable subscription name. Required when destinationType is TOPIC
commitStrategy	String	No	BATCH	Acknowledgement strategy: PER_RECORD, BATCH, or NONE
commitBatchSize	int	No	1000	Records between commits when commitStrategy is BATCH
commitIntervalMs	long	No	5000	Milliseconds between commits when commitStrategy is BATCH
receiveTimeoutMs	long	No	1000	How long a fetch blocks waiting for its first message before returning empty
maxBatchSize	int	No	500	Maximum messages returned by a single fetch

Broker Types

ActiveMQ Classic and ActiveMQ Artemis are separate products with separate JMS clients, and brokerType picks between them. Pointing the wrong client at a broker generally fails at connection time.

• CLASSIC — Apache ActiveMQ Classic (the 5.x line). Default port 61616.
• ARTEMIS — Apache ActiveMQ Artemis, also the basis of Red Hat AMQ 7 and EAP messaging.

Queues vs Topics

Queue

With destinationType: QUEUE (the default) the node creates a plain consumer on the queue. Messages are distributed among all consumers, so running multiple workers spreads the load and each message goes to one of them.

config:
  brokerUrl: "tcp://localhost:61616"
  brokerType: "CLASSIC"
  destination: "orders"
  destinationType: "QUEUE"
  encodingType: "JSON_OBJECT"

Topic

With destinationType: TOPIC the node creates a durable subscriber, which requires both clientId and subscriptionName — validation rejects the config without them.

The broker retains messages for a durable subscription while the subscriber is offline, so a restarted pipeline receives what it missed. The pair (clientId, subscriptionName) identifies the subscription to the broker, so keep both stable across restarts — changing either creates a new subscription that starts empty and leaves the old one accumulating messages on the broker.

config:
  brokerUrl: "tcp://localhost:61616"
  brokerType: "ARTEMIS"
  destination: "telemetry"
  destinationType: "TOPIC"
  clientId: "zephflow-telemetry"
  subscriptionName: "zephflow-telemetry-sub"
  encodingType: "JSON_OBJECT"

Note that a durable subscription permits only one active connection at a time, so a topic-based pipeline cannot be scaled across multiple concurrent workers using the same client id.

How It Works

The node opens a transacted JMS session and consumes in batches:

1. A fetch calls receive with receiveTimeoutMs, blocking for the first message so an idle source does not spin. Returning nothing yields an empty fetch.
2. Once a message arrives, the node drains further messages without blocking until it hits maxBatchSize or the broker has none left.
3. Each body is read — TextMessage as UTF-8 text, BytesMessage as raw bytes — and deserialized with encodingType. Any other JMS message type raises an error.
4. On commit, the JMS session transaction is committed, acknowledging every message consumed since the last commit.

Because the session is transacted, messages consumed but not yet committed are redelivered by the broker after a crash or restart. Delivery is at-least-once.

Commit strategies

• BATCH (default): commit after commitBatchSize records or commitIntervalMs milliseconds, whichever comes first.
• PER_RECORD: commit after every record. Minimizes redelivery on restart at a throughput cost.
• NONE: never commit. Messages stay unacknowledged and are redelivered — useful only for testing.

DAG Example

jobContext:
  otherProperties:
    activemq-cred:
      username: "zephflow"
      password: "supersecret"
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "activemqsource"
    config:
      brokerUrl: "tcp://activemq:61616"
      brokerType: "CLASSIC"
      destination: "orders"
      destinationType: "QUEUE"
      encodingType: "JSON_OBJECT"
      credentialId: "activemq-cred"
      commitStrategy: "BATCH"
      commitBatchSize: 1000
      commitIntervalMs: 5000
      receiveTimeoutMs: 1000
      maxBatchSize: 500
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

Related Nodes

• mqttsource: Consume messages from an MQTT broker
• kafkasource: Consume streaming data from Kafka topics
• sqssource: Read messages from an Amazon SQS queue