JMS Sink

Download connector JMS Connector for Kafka 2.1.0

This JMS Sink connector allows you to write events from Kafka to a JMS topic/queue. The connector allows you to specify the payload type sent to the JMS target:

  1. JSON
  2. AVRO
  3. MAP
  4. OBJECT

Prerequisites

  • Apache Kafka 0.11.x of above
  • Kafka Connect 0.11.x or above
  • Java 1.8
  • A JMS framework (ActiveMQ for example)
  • JMS driver for the JMS distribution - The connector does not package an client jars

Features

  1. The KCQL routing querying. Kafka topic payload field selection is supported, allowing you to select fields written to the queue or topic in JMS.
  2. Payload format selection via KCQL.
  3. Error policies for handling failures.
  4. Payload support for Schema.Struct and payload Struct, Schema.String and JSON payload and JSON payload with no schema.

KCQL Support

INSERT INTO { jms_queue|jms_topic } SELECT { FIELD, ... } FROM kafka_topic [WITHFORMAT AVRO|JSON|MAP|OBJECT] WITHTYPE { TOPIC|QUEUE }

Tip

You can specify multiple KCQL statements separated by ; to have a the connector sink multiple topics.

The JMS sink supports KCQL, Kafka Connect Query Language. The following support KCQL is available:

  1. Field selection
  2. Target queue or topic selection
  3. JMS message type selection.

Example:

-- Select all fields from topicA and write to jmsA queue
INSERT INTO jmsA SELECT * FROM topicA WITHTYPE QUEUE

-- Select 3 fields and rename from topicB and write to jmsB topic as JSON in a TextMessage
INSERT INTO jmsB SELECT x AS a, y AS b and z AS c FROM topicB WITHFORMAT JSON WITHTYPE TOPIC

JMS Topics and Queues

The sink can write to either topics or queues, specified by the WITHTYPE keyword.

JMS Payload

When a message is sent to a JMS target it can be one of the following:

  1. JSON - Send a TextMessage;
  2. AVRO - Send a BytesMessage;
  3. MAP - Send a MapMessage;
  4. OBJECT - Send an ObjectMessage

This is set by the WITHFORMAT keyword.

Payload Support

Schema.Struct and a Struct Payload

If you follow the best practice while producing the events, each message should carry its schema information. The best option is to send AVRO. Your Connector configurations options include:

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

This requires the SchemaRegistry.

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11.

Schema.String and a JSON Payload

Sometimes the producer would find it easier to just send a message with Schema.String and a JSON string. In this case your connector configuration should be set to value.converter=org.apache.kafka.connect.json.JsonConverter. This does not require the SchemaRegistry.

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11.

No schema and a JSON Payload

There are many existing systems which are publishing JSON over Kafka and bringing them in line with best practices is quite a challenge, hence we added the support. To enable this support you must change the converters in the connector configuration.

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11.

Error Polices

Landoop sink connectors support error polices. These error polices allow you to control the behavior of the sink if it encounters an error when writing records to the target system. Since Kafka retains the records, subject to the configured retention policy of the topic, the sink can ignore the error, fail the connector or attempt redelivery.

Throw

Any error on write to the target system will be propagated up and processing is stopped. This is the default behavior.

Noop

Any error on write to the target database is ignored and processing continues.

Warning

This can lead to missed errors if you do not have adequate monitoring. Data is not lost as it’s still in Kafka subject to Kafka’s retention policy. The sink currently does not distinguish between integrity constraint violations and or other exceptions thrown by any drivers or target system.

Retry

Any error on write to the target system causes the RetryIterable exception to be thrown. This causes the Kafka Connect framework to pause and replay the message. Offsets are not committed. For example, if the table is offline it will cause a write failure, the message can be replayed. With the Retry policy, the issue can be fixed without stopping the sink.

Lenses QuickStart

The easiest way to try out this is using Lenses Box the pre-configured Docker image, that comes with this connector pre-installed and ready to use. You would need to Connectors –> New Connector –> Sink –> JMS and paste your configuration

../../_images/lenses-create-jms-sink-connector.png

ActiveMQ Setup

For ActiveMQ, please visit http://activemq.apache.org/getting-started.html for the instructions of setting it up. You will also need a client jar, add this to the plugin.path of each connector worker.

Sink Connector QuickStart

Start Kafka Connect in distributed mode (see install). In this mode a Rest Endpoint on port 8083 is exposed to accept connector configurations. We developed Command Line Interface to make interacting with the Connect Rest API easier. The CLI can be found in the Stream Reactor download under the bin folder. Alternatively the Jar can be pulled from our GitHub releases page.

Installing the Connector

Connect, in production should be run in distributed mode

  1. Install and configure a Kafka Connect cluster.
  2. Create a folder on each server called plugins/lib.
  3. Copy into the above folder the required connector jars from the stream reactor download.
  4. Edit connect-avro-distributed.properties in the etc/schema-registry folder and uncomment the plugin.path. option. Set it to the root directory i.e. plugins you deployed the stream reactor connector jars in step 2.
  5. Start Connect, bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties.

Connect Workers are long running processes so set an init.d or systemctl service accordingly.

Starting the Connector (Distributed)

Download, and install Stream Reactor. Follow the instructions here if you have not already done so. All paths in the quickstart are based on the location you installed the Stream Reactor.

Once the Connect has started we can now use the kafka-connect-tools cli to post in our distributed properties file for JMS. For the CLI to work including when using the dockers you will have to set the following environment variable to point the Kafka Connect Rest API.

export KAFKA_CONNECT_REST="http://myserver:myport"

➜  bin/connect-cli create jms-sink < conf/jms-sink.properties

name=jms-sink
connector.class=com.datamountaineer.streamreactor.connect.jms.sink.JMSSinkConnector
tasks.max=1
topics=topic
connect.jms.url=tcp://localhost:61616
connect.jms.initial.context.factory=org.apache.activemq.jndi.ActiveMQInitialContextFactory
connect.jms.connection.factory=ConnectionFactory
connect.jms.queues=jms-queue
connect.jms.kcql=INSERT INTO jms-queue SELECT * FROM topic

If you switch back to the terminal you started Kafka Connect in, you should see the Elastic Search Sink being accepted and the task starting.

We can use the CLI to check if the connector is up but you should be able to see this in logs as well.

#check for running connectors with the CLI
➜ bin/connect-cli ps
jms-sink

Test Records

Tip

If your input topic does not match the target use Lenses SQL to transform in real-time the input, no Java or Scala is required!

Now we need to put some records it to the jms_test topics. We can use the kafka-avro-console-producer to do this. Start the producer and pass in a schema to register in the Schema Registry. The schema has an id field of type int and a random_field of type string.

bin/kafka-avro-console-producer \
 --broker-list localhost:9092 --topic jms_test \
 --property value.schema='{"type":"record","name":"User",
"fields":[{"name":"firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"salary","type":"double"}]}'

Now the producer is waiting for input. Paste in the following:

{"firstName": "John", "lastName": "Smith", "age":30, "salary": 4830}
{"firstName": "Anna", "lastName": "Jones", "age":28, "salary": 5430}

Now check for records in ActiveMQ.

Configurations

The Kafka Connect framework requires the following in addition to any connectors specific configurations:

Config Description Type Value
name Name of the connector string This must be unique across the Connect cluster
topics
The topics to sink.
The connector will check that this matches the KCQL statement
 string  
tasks.max The number of tasks to scale output int 1
connector.class Name of the connector class string com.datamountaineer.streamreactor.connect.jms.sink.JMSSinkConnector

Connector Configurations

Config Description Type
connect.jms.url Provides the JMS broker url string
connect.jms.username Provides the user for the JMS connection string
connect.jms.password Provides the password for the JMS connection string
connect.jms.initial.context.factory
Initial Context Factory,
e.g: org.apache.activemq.jndi.ActiveMQInitialContextFactory
 string
connect.jms.connection.factory The ConnectionFactory implementation to use string
connect.jms.kcql
KCQL expression describing
field selection and routes. The
kcql expression also handles setting
the JMS destination type, i.e. TOPIC or
QUEUE via the withtype keyword
 string

Optional Configurations

Config Description Type Default
connect.jms.destination.selector
Selector to use for destination lookup.
Either CDI or JNDI
 string CDI
connect.jms.initial.context.extra.params
List (comma separated) of extra
properties as key/value pairs using = as a delimiter
to supply to the initial context e.g.
SOLACE_JMS_VPN=my_solace_vp
 string  
connect.jms.error.policy
Specifies the action to be
taken if an error occurs while inserting the data.
There are three available options, NOOP, the error
is swallowed, THROW, the error is allowed
to propagate and retry.
For RETRY the Kafka message is redelivered up
to a maximum number of times specified by the
connect.jms.max.retries option
 string THROW
connect.jms.max.retries
The maximum number of times a message
is retried. Only valid when the
connect.jms.error.policy is set to RETRY
 string 10
connect.jms.retry.interval
The interval, in milliseconds between retries,
if the sink is using
connect.jms.error.policy set to RETRY
 string 60000
connect.progress.enabled
Enables the output for how many
records have been processed
 boolean false
connect.jms.polling.timeout
Provides the timeout to poll incoming messages |
 long 1000
connect.jms.headers
Collection of static JMS headers included.
Separated by question mark
 string false

Schema Evolution

Not applicable.

Kubernetes

Helm Charts are provided at our repo, add the repo to your Helm instance and install. We recommend using the Landscaper to manage Helm Values since typically each Connector instance has its own deployment.

Add the Helm charts to your Helm instance:

helm repo add landoop https://landoop.github.io/kafka-helm-charts/

TroubleShooting

Please review the FAQs and join our slack channel