<a id="activemq-sink-connector"></a>

# ActiveMQ Sink Connector for Confluent Platform

The Kafka Connect ActiveMQ Sink Connector is used to move messages from
Apache Kafka® to an [ActiveMQ](https://activemq.apache.org/) cluster.

#### NOTE
- If you need to use the Java Naming and Directory Interface™ (JNDI)
  to connect to ActiveMQ, there is a general [JMS Sink Connector for Confluent
  Platform](https://docs.confluent.io/kafka-connect-jms-sink/current/)
  available that uses a JNDI-based mechanism to connect to the JMS broker.
- <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

  Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
  [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
  page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
  are no longer available for download.

ActiveMQ Artemis is supported with the ActiveMQ Sink connector.

## Features

The ActiveMQ Sink connector includes the following features:

- [At least once delivery](#activemq-sink-at-least-once)
- [Dead Letter Queue](#activemq-sink-dead-letter-queue)
- [Multiple tasks](#activemq-sink-multiple-tasks)

<a id="activemq-sink-at-least-once"></a>

### At least once delivery

This connector guarantees that records are delivered at least once from the
Kafka topic.

<a id="activemq-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. For
information about accessing and using the DLQ, see [Confluent Platform
Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="activemq-sink-multiple-tasks"></a>

### Multiple tasks

The ActiveMQ Sink connector supports running one or more tasks. You can specify
the number of tasks in the `tasks.max` configuration parameter. This can lead
to performance gains when multiple files need to be parsed.

## Install the ActiveMQ Sink Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later
- Connect: Confluent Platform 4.1.0 or later, or Kafka 1.1.0 or later (requires header
  support in Connect)
- [ActiveMQ 5.x](https://activemq.apache.org/components/classic)
- Java 1.8

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-activemq-sink:latest
```

You can install a specific version by replacing `latest` with a version number
as shown in the following example:

```bash
confluent connect plugin install confluentinc/kafka-connect-activemq-sink:2.2.0
```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-activemq-sink) for
your connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/install.html).

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](connector_config.md#activemq-sink-connector-license-config) for license properties and
[License topic configuration](connector_config.md#activemq-sink-license-topic-configuration) for information about the
license topic.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for ActiveMQ Sink Connector for Confluent Platform](connector_config.md#activemq-sink-connector-config).

## Client Library JARs

The Kafka Connect ActiveMQ connector includes all of the client libraries
required to work with ActiveMQ.

#### NOTE
The ActiveMQ Sink Connector uses the `org.apache.activemq:activemq-client:5.14.4` client library.

## Quick Start

This quick start uses the ActiveMQ Sink Connector to consume records from Kafka
and send them to an ActiveMQ broker.

1. [Install ActiveMQ](https://activemq.apache.org/getting-started#installation-procedure-for-unix)
2. [Start ActiveMQ](https://activemq.apache.org/getting-started#starting-activemq)
3. Install the connector through the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html).
   ```bash
   # run from your Confluent Platform installation directory
   confluent connect plugin install confluentinc/kafka-connect-activemq-sink:latest
   ```
4. Start Confluent Platform.
   ```bash
   confluent local start
   ```
5. [Produce](https://docs.confluent.io/current/cli/command-reference/confluent-produce.html)
   test data to the `sink-messages` topic in Kafka.
   ```bash
   seq 10 | confluent local produce sink-messages
   ```
6. Create a `activemq-sink.json` file with the following contents:
   ```json
   {
     "name": "AMQSinkConnector",
     "config": {
       "connector.class": "io.confluent.connect.jms.ActiveMqSinkConnector",
       "tasks.max": "1",
       "topics": "sink-messages",
       "activemq.url": "tcp://localhost:61616",
       "activemq.username": "connectuser",
       "activemq.password": "connectuser",
       "jms.destination.type": "queue",
       "jms.destination.name": "connector-quickstart",
       "key.converter": "org.apache.kafka.connect.storage.StringConverter",
       "value.converter": "org.apache.kafka.connect.storage.StringConverter",
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "confluent.topic.replication.factor": "1"
     }
   }
   ```
7. Load the ActiveMQ Sink Connector.
   ```bash
   confluent local load jms --config activemq-sink.json
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.
8. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status AMQSinkConnector
   ```
9. Navigate to the [ActiveMQ Admin UI](http://localhost:8161/admin) or use
   the following ActiveMQ CLI command to confirm the messages were delivered to the
   `connector-quickstart` queue.
   ```bash
   ./bin/activemq consumer --destination connector-quickstart --messageCount 10
   ```

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## JMS Message Formats

The format of outgoing JMS Message values is configured with the
`jms.message.format` property, using one of the following options:

### string (*default*)

When using the `string` Message Format, record values are run through
`Values.convertToString(...)` from the Connect Data package and produced as a
JMS `TextMessage`.

Primitive values are converted to their String equivalent and structured objects
are transformed to a sensible string representation that is similar to the JSON
representation, with the exception of simple string values (not in objects or
arrays) which are unquoted.

### avro

Record values are serialized without the Avro schema information and produced as
a JMS `BytesMessage`. JMS consumers **must have the schema** to deserialize
the data.

#### IMPORTANT
The connector attempts to infer the Avro schema for records that have no
schema. If the connector cannot infer the schema, the task is killed. If you
are processing data without a schema, consider using one of the other
`jms.message.format` configurations.

### json

Record values are converted to a UTF-8 encoded JSON representation and produced
as a JMS `TextMessage`.

### bytes

Record values are passed along in bytes form without any conversion.

#### IMPORTANT
Record values must be converted to bytes form before the connector processes
them. Configure the `value.converter` property to
`org.apache.kafka.connect.converters.ByteArrayConverter` to ensure that
record values arrive in byte format.

## Forwarding Kafka Properties to JMS

The connector can be configured to forward various values from the Kafka record
to the JMS Message.

* Enable `jms.forward.kafka.key` to convert the record’s key to a String and
  forward it as the `JMSCorrelationID`.
* Enable `jms.forward.kafka.metadata` to forward the record’s topic,
  partition, and offset on JMS Message properties.
  * Kafka topic is applied to the message as a String property named `KAFKA_TOPIC`.
  * Partition is applied to the message as an Int property named `KAFKA_PARTITION`.
  * Offset is applied to the message as a Long property named `KAFKA_OFFSET`.
* Enable `jms.forward.kafka.headers` to add each header from the SinkRecord to
  the JMS Message as a String property.

#### NOTE
The connector converts the record key and headers to a sensible string
representation that is similar to the JSON representation, with the
exception of simple string values (not in objects or arrays) which are
unquoted. No other conversion is done to the key and headers before
forwarding them on the JMS Message. If another format is needed,
out-of-the-box or custom [Single Message
Transformation](/platform/current/connect/transforms/index.html) can be used with the connector
to transform the record keys and/or headers to the desired string
representation before the JMS sink connector processes each record.