Important

You are viewing documentation for an older version of Confluent Platform. For the latest, click here.

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	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	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`.