.. _ibmmq-source-connector:
|kconnect-long| IBM MQ Source Connector
=======================================
The IBM MQ Source Connector is used to read messages from an
`IBM MQ <https://www-03.ibm.com/software/products/en/ibm-mq>`_
cluster and write them to a |ak-tm| topic.
.. note::
Confluent Platform also includes a general
:ref:`JMS source connector <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 instead.
Install the IBM MQ Source Connector
-----------------------------------
.. include:: ../includes/connector-native-install-cpe.rst
.. include:: ../includes/connector-install-hub.rst
.. codewithvars:: bash
confluent-hub install confluentinc/kafka-connect-ibmmq:latest
.. include:: ../includes/connector-install-version.rst
.. codewithvars:: bash
confluent-hub install confluentinc/kafka-connect-ibmmq:|release|
--------------------------
Install Connector Manually
--------------------------
`Download and extract the ZIP file <https://www.confluent.io/connector/kafka-connect-ibm-mq/#download>`_ for your
connector and then follow the manual connector installation :ref:`instructions <connect_install_connectors>`.
.. _ibmmq-source-connector-license-key:
License
-------
.. include:: ../includes/enterprise-license.rst
See :ref:`ibmmq-source-connector-license-config` for license properties and :ref:`ibmmq-source-license-topic-configuration` for information about the license topic.
.. _ibmmq-source-connector-client-libs:
Client Libraries
----------------
To use the |kconnect-long| IBM MQ connector, you **must** download the IBM MQ client library JAR files. Complete the following steps to get these JAR files.
.. note::
The exact list of JAR files from the IBM client library may vary depending on the Java software version.
#. Follow the instructions at the IBM Support page `Getting IBM MQ classes for Java and JMS <https://www.ibm.com/support/pages/getting-ibm-mq-classes-java-and-jms>`__. Once completed, you should have a downloaded file named ``9.0.0.7-IBM-MQ-Install-Java-All.jar`` (your version number may be different).
#. To extract the JAR files, enter the following command:
::
java -jar 9.0.0.7-IBM-MQ-Install-Java-All.jar
Before you can use, extract, or install IBM MQ V9.0, you must
accept the terms of 1. IBM International License Agreement for
Evaluation of Programs 2. IBM International Program License Agreement
and additional license information. Please read the following license
agreements carefully.
...omitted
Target directory for product files? /tmp/ibmmq-jars/
Extracting files to /tmp/ibmmq-jars/wmq
Successfully extracted all product files.
#. Copy all of the required JAR files to the same folder as the kafka-connect-ibmmq JAR file and dependencies.
::
cp /tmp/ibmmq-jars/wmq/JavaSE/* /u01/connectors/kafka-connect-ibmmq/
.. important::
These JAR files need to be copied for each of your |cp| installations where you want to run this connector.
#. List the JAR files to verify that they were copied successfully.
::
ls -l /u01/connectors/kafka-connect-ibmmq/
Make sure that you have the following files:
* com.ibm.mq.allclient.jar
* com.ibm.mq.traceControl.jar
* jms.jar
.. _ibmmq-source-connector-messagetypes:
JMS Message types
-----------------
The connector currently supports only
`TextMessage <http://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html>`_
and `BytesMessage <http://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html>`_
but does not currently support `ObjectMessage <http://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html>`_
or `StreamMessage <http://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html>`_
.
.. _ibmmq-source-connection:
Connecting to IBM MQ
--------------------
Before you can use this connector, you must install the IBM MQ client JARs into this connector's installation directory. See the IBM MQ documentation for details.
This connector connects directly to the IBM MQ using a number of configuration properties that should match your environment.
The following example shows a typical configuration of the connector for use with
:ref:`distributed mode <distributed-workers>`:
.. codewithvars:: bash
{
"name": "connector1",
"config": {
"connector.class": "io.confluent.connect.ibm.mq.IbmMQSourceConnector",
"kafka.topic":"MyKafkaTopicName",
"mq.hostname":"localhost",
"mq.port":"61616",
"mq.transport.type":"client",
"mq.queue.manager":"QMA",
"mq.channel":"SYSTEM.DEF.SVRCONN",
"jms.destination.name":"testing",
"jms.destination.type":"queue",
"confluent.license":"",
"confluent.topic.bootstrap.servers":"localhost:9092"
}
}
The connector supports other :ref:`configuration options <ibmmq-source-connector-config>` not included in the example above.
.. _ibmmq-source-connector-topics:
Topics
------
This connector consumes messages from IBM MQ using the configured
:ref:`message selectors <ibmmq-source-connector-config>` and writes them to a single |ak| topic.
If you want to write messages to multiple topics, use a simple message transform that routes the messages based upon your criteria.
.. _ibmmq-source-connector-schemas:
Schemas
-------
The IBM MQ connector produces messages with keys and values that adhere to the schemas described in the following sections.
.. _ibmmq-source-connector-schema-key:
----------------------------
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() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSMessageID()>`_. |
+-----------+--------+----------+---------------+---------------------------------------------------------------------------------------------------------------------------------------------+
.. _ibmmq-source-connector-schema-value:
------------------------------
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() <http://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 subinterfaces of `Message <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html>`_. `BytesMessage <http://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html>`_ = `bytes`, `MapMessage <http://docs.oracle.com/javaee/6/api/javax/jms/MapMessage.html>`_ = `map`, `ObjectMessage <http://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html>`_ = `object`, `StreamMessage <http://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html>`_ = `stream` and `TextMessage <http://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html>`_ = `text`. The corresponding field will be populated with the values from the respective Message subinterface. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| timestamp | INT64 | yes | | Data from the `getJMSTimestamp() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSTimestamp()>`_ method. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| deliveryMode | INT32 | yes | | This field stores the value of `Message.getJMSDeliveryMode() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSDeliveryMode()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| correlationID | STRING | no | | This field stores the value of `Message.getJMSCorrelationID() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSCorrelationID()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| replyTo | :ref:`Destination <ibmmq-source-connector-schema-destination>` | no | | This schema is used to represent a JMS Destination, and is either `queue <http://docs.oracle.com/javaee/6/api/javax/jms/Queue.html>`_ or `topic <http://docs.oracle.com/javaee/6/api/javax/jms/Topic.html>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| destination | :ref:`Destination <ibmmq-source-connector-schema-destination>` | no | | This schema is used to represent a JMS Destination, and is either `queue <http://docs.oracle.com/javaee/6/api/javax/jms/Queue.html>`_ or `topic <http://docs.oracle.com/javaee/6/api/javax/jms/Topic.html>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| redelivered | BOOLEAN | yes | | This field stores the value of `Message.getJMSRedelivered() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSRedelivered()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| type | STRING | no | | This field stores the value of `Message.getJMSType() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSType()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| expiration | INT64 | yes | | This field stores the value of `Message.getJMSExpiration() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSExpiration()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| priority | INT32 | yes | | This field stores the value of `Message.getJMSPriority() <http://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSPriority()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| properties | Map of STRING, :ref:`PropertyValue <ibmmq-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[]) <http://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html#readBytes(byte[])>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| map | Map of STRING, :ref:`PropertyValue <ibmmq-source-connector-schema-propertyvalue>` | no | | This field stores the data from all of the map entries returned from `MapMessage.getMapNames() <http://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() <http://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html#getText()>`_. |
+---------------+-----------------------------------------------------------------------------------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
.. _ibmmq-source-connector-schema-destination:
------------------------------------
io.confluent.connect.jms.Destination
------------------------------------
This schema is used to represent a JMS Destination, and is either `queue <http://docs.oracle.com/javaee/6/api/javax/jms/Queue.html>`_ or `topic <http://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() <http://docs.oracle.com/javaee/6/api/javax/jms/Queue.html#getQueueName()>`_ or `Topic.getTopicName() <http://docs.oracle.com/javaee/6/api/javax/jms/Topic.html#getTopicName()>`_. |
+-----------------+--------+----------+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
.. _ibmmq-source-connector-schema-propertyvalue:
--------------------------------------
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() <http://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``. |
+--------------+---------+----------+---------------+-----------------------------------------------------------------------------------------------------------------------------------------------------+
Additional documentation
------------------------
.. toctree::
:maxdepth: 1
source_connector_config
examples
changelog