ActiveMQ Source Connector

The ActiveMQ Source Connector is used to read messages from an ActiveMQ cluster and write them to a Kafka topic.

Note

Confluent Platform also includes a general JMS source connector that uses a JNDI-based mechanism to connect to the JMS broker. If you have to use JNDI to connect to your JMS broker, consider using that connector.

Client Libraries

The Confluent ActiveMQ connector includes all the libraries required to work with ActiveMQ, so there is nothing else to install.

License Key

Without the license key, you can use the ActiveMQ connector for a 30-day trial period. Simply remove the confluent.license from your connector configuration file or leave that property blank. See the configuration options for more details.

If you are a Confluent customer, you can contact customer support and ask for an ActiveMQ Connector license key. Then add confluent.license configuration to the ActiveMQ Connector configuration file (see below) followed by the key you received from Confluent support.

JMS Message types

The connector currently supports only TextMessage and BytesMessage but does not currently support ObjectMessage or StreamMessage .

Connecting to ActiveMQ

This connector connects directly to ActiveMQ using a connection URL 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:

{
  "name": "connector1",
  "config": {
    "connector.class": "io.confluent.connect.activemq.ActiveMQSourceConnectorConfig",
    "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 not included in the example above.

Topics

This connector consumes messages from ActiveMQ using the configured message selectors 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().

io.confluent.connect.jms.Value

This schema is used to store the value of the JMS message.

The schema defines the following fields:

Name Schema Required Default Value Documentation
messageID STRING yes   This field stores the value of Message.getJMSMessageID().
messageType STRING yes   This field stores the type of message that was received. This corresponds to the subinterfaces of Message. BytesMessage = bytes, MapMessage = map, ObjectMessage = object, StreamMessage = stream and TextMessage = text. The corresponding field will be populated with the values from the respective Message subinterface.
timestamp INT64 yes   Data from the getJMSTimestamp() method.
deliveryMode INT32 yes   This field stores the value of Message.getJMSDeliveryMode().
correlationID STRING no   This field stores the value of Message.getJMSCorrelationID().
replyTo Destination no   This schema is used to represent a JMS Destination, and is either queue or topic.
destination Destination no   This schema is used to represent a JMS Destination, and is either queue or topic.
redelivered BOOLEAN yes   This field stores the value of Message.getJMSRedelivered().
type STRING no   This field stores the value of Message.getJMSType().
expiration INT64 yes   This field stores the value of Message.getJMSExpiration().
priority INT32 yes   This field stores the value of Message.getJMSPriority().
properties Map of STRING, 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[]).
map Map of STRING, PropertyValue no   This field stores the data from all of the map entries returned from MapMessage.getMapNames() for the Message indexed by their key.
text STRING no   This field stores the value from TextMessage.html.getText().

io.confluent.connect.jms.Destination

This schema is used to represent a JMS Destination, and is either queue or topic.

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() or Topic.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().

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.