# ActiveMQ Source Connector for Confluent Platform #### NOTE - The [JMS Source Connector for Confluent Platform](https://docs.confluent.io/kafka-connect-jms-source/current/index.html) is available for download from Confluent Marketplace. This connector uses a Java Naming and Directory Interface (JNDI)-based mechanism to connect to the JMS broker. If you need to use JNDI to connect to your JMS broker, consider using that connector instead. - 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. The Kafka Connect ActiveMQ Source connector is used to read messages from an [ActiveMQ](https://activemq.apache.org/) cluster and write them to an Apache Kafka® topic. ActiveMQ Artemis is supported with the ActiveMQ Source connector. ## Features The ActiveMQ Source connector includes the following features: - [At least once delivery](#activemq-source-at-least-once-delivery) - [Multiple tasks](#activemq-source-multiple-tasks) - [Client-side encryption](#activemq-sink-csfle-sm) ### At least once delivery This connector guarantees that records are delivered at least once to the Kafka topic. If the connector restarts, there may be some duplicate records in the Kafka topic. ### Multiple tasks The ActiveMQ Source 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. ### Client-side encryption This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html). ## Install the ActiveMQ 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. - An install of the latest (`latest`) connector version. To install the `latest` connector version, navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install confluentinc/kafka-connect-activemq: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:13.1.0 ``` ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-activemq) 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](source_connector_config.md#activemq-source-connector-license-config) for license properties and [License topic configuration](source_connector_config.md#activemq-source-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 Source Connector for Confluent Platform](source_connector_config.md#activemq-source-connector-config). To understand how the connector internally configures the acknowledgement mode, see the following section. ### Acknowledgement mode The connector internally uses `CLIENT_ACKNOWLEDGE` mode to receive and acknowledge messages from the JMS broker. In this mode, acknowledging any message will acknowledge every message received (see section 6.2.10 in the [JMS 2.0 Specification](https://download.oracle.com/otndocs/jcp/jms-2_0-fr-eval-spec/)). To prevent messages from being prematurely acknowledged, the connector processes only one message at time. In other words, the connector will not try to receive new messages until the last message is committed to a Kafka topic. This might compromise the throughput of the connector, but messages will be transferred to Kafka successfully. ## Client Libraries The Kafka Connect ActiveMQ connector includes all the libraries required to work with ActiveMQ, so there is nothing else to install. ## JMS Message types The connector currently supports only [TextMessage](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html) and [BytesMessage](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html) but does not currently support [ObjectMessage](https://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html) or [StreamMessage](https://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html) . ## Connecting to ActiveMQ This connector connects directly to ActiveMQ using a [connection URL](https://activemq.apache.org/connection-configuration-uri.html) for your messaging system, using the ActiveMQ client libraries included with the connector. The following example shows a typical configuration of the connector for use with [distributed mode](/platform/current/connect/concepts.html#distributed-workers): ```bash { "name": "connector1", "config": { "connector.class": "io.confluent.connect.activemq.ActiveMQSourceConnector", "kafka.topic":"MyKafkaTopicName", "activemq.url":"tcp://localhost:61616", "jms.destination.name":"testing", "jms.destination.type":"queue", "confluent.license":"", "confluent.topic.bootstrap.servers":"localhost:9092" } } ``` The connector supports other [configuration options](source_connector_config.md#activemq-source-connector-config) not included in the previous example. 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). ## Topics This connector consumes messages from ActiveMQ using the configured [message selectors](source_connector_config.md#activemq-source-connector-config) and writes them to a single Kafka topic. If you want to write messages to multiple topics, use a simple message transform that routes the messages based upon your criteria. ## Schemas The ActiveMQ Connector produces messages with keys and values that adhere to the schemas described in the following sections. ### io.confluent.connect.jms.Key This schema is used to store the incoming MessageID on the message interface. This will ensure that when that if the same message id arrives it will end up in the same partition. In practice this should never occur. The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |-----------|----------|------------|-----------------|------------------------------------------------------------------------------------------------------------------------------------------------| | messageID | STRING | yes | | This field stores the value of
[Message.getJMSMessageID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSMessageID()). | ### io.confluent.connect.jms.Value This schema is used to store the value of the JMS message. The schema defines the following fields: | Name | Schemna | Required | Default Value | Documentation | |---------------|---------------------------------------------------------------------------------|------------|-----------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | messageID | STRING | yes | | This field stores the value of
[Message.getJMSMessageID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSMessageID()). | | messageType | STRING | yes | | This field stores the type of message that was received. This corresponds
to the sub-interfaces of [Message](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html).
[BytesMessage](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html) =
`bytes`, [MapMessage](https://docs.oracle.com/javaee/6/api/javax/jms/MapMessage.html) = `map`,
[ObjectMessage](https://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html) =
`object`, [StreamMessage](https://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html) =
`stream` and [TextMessage](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html) =
`text`. The corresponding field will be populated with the values from the
respective message sub-interface. | | timestamp | INT64 | yes | | Data from the [getJMSTimestamp()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSTimestamp())
method. | | deliveryMode | INT32 | yes | | This field stores the value of [Message.getJMSDeliveryMode()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSDeliveryMode()).
method. | | correlationID | STRING | no | | This field stores the value of [Message.getJMSCorrelationID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSCorrelationID()).
method. | | replyTo | [Destination](#activemq-source-connector-schema-destination) | no | | This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). | | destination | [Destination](#activemq-source-connector-schema-destination) | no | | This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). | | redelivered | BOOLEAN | yes | | This field stores the value of [Message.getJMSRedelivered()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSRedelivered()). | | type | STRING | no | | This field stores the value of [Message.getJMSType()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSType()). | | expiration | INT64 | no | | This field stores the value of [Message.getJMSExpiration()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSExpiration()). | | priority | INT32 | no | | This field stores the value of [Message.getJMSPriority()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSPriority()). | | properties | Map of STRING, [PropertyValue](#activemq-source-connector-schema-propertyvalue) | yes | | This field stores the data from all of the properties for the Message indexed by their propertyName. | | bytes | BYTES | no | | This field stores the value from
[BytesMessage.html.readBytes(byte[])](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html#readBytes(byte[])). | | map | Map of STRING, [PropertyValue](#activemq-source-connector-schema-propertyvalue) | no | | This field stores the data from all of the map entries returned from
[MapMessage.getMapNames()](https://docs.oracle.com/javaee/6/api/javax/jms/MapMessage.html#getMapNames())
for the Message indexed by their key. | | text | STRING | no | | This field stores the value from [TextMessage.html.getText()](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html#getText()). | ### io.confluent.connect.jms.Destination This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |-----------------|----------|------------|-----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | destinationType | STRING | yes | | The type of JMS Destination, and either `queue` or `topic`. | | name | STRING | yes | | The name of the destination. This will be the value of
[Queue.getQueueName()](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html#getQueueName())
or [Topic.getTopicName()](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html#getTopicName()). | ### io.confluent.connect.jms.PropertyValue This schema is used to store the data that is found in the properties of the message. To ensure that the proper type mappings are preserved field `propertyType` stores the value type for the field. The corresponding field in the schema will contain the data for the property. This ensures that the data is retrievable as the type returned by [Message.getObjectProperty()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getObjectProperty(java.lang.String)). The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |--------------|----------|------------|-----------------|-----------------------------------------------------------------------------------------------------------------------------------------| | propertyType | STRING | yes | | The Java type of the property on the Message. One of `boolean`, `byte`,
`short`, `integer`, `long`, `float`, `double`, or `string`. | | boolean | BOOLEAN | no | | The value stored as a boolean. Null unless `propertyType` is set to `boolean`. | | byte | INT8 | no | | The value stored as a byte. Null unless `propertyType` is set to `byte`. | | short | INT16 | no | | The value stored as a short. Null unless `propertyType` is set to `short`. | | integer | INT32 | no | | The value stored as a integer. Null unless `propertyType` is set to `integer`. | | long | INT64 | no | | The value stored as a long. Null unless `propertyType` is set to `long`. | | float | FLOAT32 | no | | The value stored as a float. Null unless `propertyType` is set to `float`. | | double | FLOAT64 | no | | The value stored as a double. Null unless `propertyType` is set to `double`. | | string | STRING | no | | The value stored as a string. Null unless `propertyType` is set to `string`. |