Migrating the IBM MQ source connector

After you migrate your Kafka Connect cluster to Confluent Platform, you can migrate the IBM MQ source connector from Event Streams to Confluent for Kubernetes (CFK). The connector reads messages from an IBM MQ queue and writes them to a Kafka topic.

For general connector migration steps including how to stop the connector, extract the offset, and set the starting position on Confluent, see migrating connectors.

Migration paths

To migrate the IBM MQ source connector to Confluent Platform, you have the following two options:

Note: The Confluent IBM MQ source connector is built and supported by Confluent, and is not the IBM open-source connector repackaged. If you keep the IBM open-source connector, support for the connector code comes from IBM only and does not extend to its integration with Confluent Platform.

Option A: Confluent IBM MQ source connector

This section covers how to install and configure the Confluent-native IBM MQ source connector.

Plugin packaging

The Confluent IBM MQ source connector is available on Confluent Hub as confluentinc/kafka-connect-ibmmq. It is a Confluent commercial connector that requires a Confluent Platform subscription after a 30-day evaluation period. For the full configuration reference, see Confluent Hub: kafka-connect-ibmmq, the connector overview, and the configuration reference.

The connector does not bundle the IBM MQ client library, and does not start without it. To obtain com.ibm.mq.allclient.jar, follow the instructions in download the IBM MQ client library JAR files, then copy the file into the connector lib directory as shown in the following steps.

Complete the following steps to package the connector:

1. Add the following to your Dockerfile to install the connector and copy the IBM MQ client library:

   FROM confluentinc/cp-server-connect:<cp-version>
   
   # Install the Confluent IBM MQ source connector
   RUN confluent-hub install --no-prompt confluentinc/kafka-connect-ibmmq:<version>
   
   # Copy the IBM MQ client library, which is required but not bundled in the Confluent Hub package.
   # Obtain com.ibm.mq.allclient.jar from your IBM MQ installation: <MQ_INSTALL>/java/lib/
   COPY com.ibm.mq.allclient.jar \
        /usr/share/confluent-hub-components/confluentinc-kafka-connect-ibmmq/lib/
   

   Where:

   • <cp-version> is the Confluent Platform version that matches your target deployment, available on Docker Hub: confluentinc/cp-server-connect.
   • <version> is your required connector version, available on Confluent Hub: kafka-connect-ibmmq.

2. The connector requires a Confluent license at runtime. Add the following to your Connect custom resource configOverrides:

   configOverrides:
     server:
       - confluent.license=<your-license-key>
       - confluent.topic.bootstrap.servers=<kafka-bootstrap-endpoint>
   

   For a 30-day evaluation, leave confluent.license empty.

Configuration mapping from Event Streams to Confluent connector

The Confluent IBM MQ source connector uses different configuration field names. Review the full mapping before migrating.

Connection and queue settings:

Message handling:

TLS settings:

The Confluent connector uses mq.tls.* keys instead of mq.ssl.*.

Example Connector custom resource

The following example shows a CFK Connector custom resource for the Confluent IBM MQ source connector.

Note: All values in spec.configs must be strings. Boolean and numeric values must be quoted as strings.

apiVersion: platform.confluent.io/v1beta1
kind: Connector
metadata:
  name: mq-source-connector
  namespace: <cp-namespace>
spec:
  class: io.confluent.connect.ibm.mq.IbmMQSourceConnector
  taskMax: 1
  connectClusterRef:
    name: connect
  configs:
    # MQ connection
    mq.hostname: "<mq-host>"
    mq.port: "1414"
    mq.queue.manager: "QM1"
    mq.channel: "KAFKA.CONN.SVRCONN"
    mq.username: "app"
    mq.password: "<password>"
    mq.transport.type: "client"

    # Source queue
    jms.destination.name: "KAFKA.SOURCE.QUEUE"
    jms.destination.type: "queue"

    # Kafka target topic
    kafka.topic: "MY.KAFKA.TOPIC"

    # Optional: message filtering
    # jms.message.selector: "JMSType = 'OrderMessage'"

    key.converter: "org.apache.kafka.connect.storage.StringConverter"
    value.converter: "org.apache.kafka.connect.storage.StringConverter"

Option B: Same open-source IBM MQ connector JAR files

This section covers how to package the same open-source IBM MQ connector JAR files from Event Streams into your plugin image.

Plugin packaging

Extract the IBM MQ source connector JAR files from your existing Event Streams Connect image, or download the open-source release from GitHub: ibm-messaging/kafka-connect-mq-source.

Complete the following steps to package the connector:

1. Add the following to your Dockerfile to copy the JAR files into the plugin image:

   FROM confluentinc/cp-server-connect:<cp-version>
   
   # Each connector must be in its own subdirectory under plugin.path.
   COPY kafka-connect-mq-source-<version>-jar-with-dependencies.jar \
        /opt/kafka/plugins/mq-source/
   

2. Ensure the Connect custom resource includes /opt/kafka/plugins in plugin.path:

   configOverrides:
     server:
       - plugin.path=/usr/share/java,/usr/share/confluent-hub-components,/opt/kafka/plugins
   

Configuration compatibility

All configuration keys are identical to Event Streams. The only changes are the CFK custom resource structure and that all values must be quoted as strings. For the full configuration reference, see GitHub: ibm-messaging/kafka-connect-mq-source.

The IBM open-source connector uses mq.ssl.* fields for TLS configuration:

Example Connector custom resource

The following example shows a CFK Connector custom resource for the IBM open-source MQ source connector:

apiVersion: platform.confluent.io/v1beta1
kind: Connector
metadata:
  name: mq-source-connector   # must match the name used on Event Streams for offset continuity
  namespace: <cp-namespace>
spec:
  class: com.ibm.eventstreams.connect.mqsource.MQSourceConnector
  taskMax: 1
  configs:
    # MQ connection
    mq.queue.manager: "QM1"
    mq.connection.name.list: "<mq-host>(1414)"
    mq.channel.name: "KAFKA.CONN.SVRCONN"
    mq.queue: "KAFKA.SOURCE.QUEUE"
    mq.user.name: "app"
    mq.password: "<password>"
    mq.user.authentication.mqcsp: "true"

    # Message handling
    mq.message.body.jms: "true"
    mq.record.builder: "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder"
    mq.jms.properties.copy.to.kafka.headers: "true"
    mq.message.mqmd.read: "true"
    mq.record.builder.key.header: "JMSCorrelationID"

    # Polling
    mq.batch.size: "250"
    mq.message.receive.timeout: "2000"
    mq.max.poll.blocked.time.ms: "2000"

    # Reconnection
    mq.client.reconnect.options: "ASDEF"
    mq.reconnect.delay.min.ms: "64"
    mq.reconnect.delay.max.ms: "8192"

    # Kafka target
    topic: "MY.KAFKA.TOPIC"
    key.converter: "org.apache.kafka.connect.storage.StringConverter"
    value.converter: "org.apache.kafka.connect.storage.StringConverter"

  connectClusterRef:
    name: connect

Offset handling

Offset handling depends on which migration option you choose and which delivery mode you use. For steps about how to extract and inject offsets, see connector offset handling.

Option A: Confluent connector offset behavior

The Confluent IBM MQ source connector (io.confluent.connect.ibm.mq.IbmMQSourceConnector) delivers messages at least once by default, and does not record a resume position in the Kafka Connect offset topic. Its optional exactly-once mode tracks progress in a state topic that is created in IBM MQ, not in the Connect offset topic. For more information, see delivery guarantee in the Confluent documentation.

There is therefore no offset to carry across from Event Streams. Point the Confluent connector at the same MQ queue. It reads whatever messages remain on the queue when the Event Streams connector was stopped, including any messages that the Event Streams connector read but did not confirm as committed to Kafka.

Important: Confirm that all downstream consumers have consumed all messages produced by the Event Streams connector before starting the Confluent IBM MQ source connector.

Option B: Open-source connector offset behavior

Offset handling when using the open-source IBM MQ source connector depends on the delivery mode:

• Standard at-least-once delivery: The connector relies on MQ transactional behavior. Messages are consumed in a JMS transaction and removed from the MQ queue only after Kafka acknowledges receipt. On restart, the connector reads from the beginning of the MQ queue, and any uncommitted messages remain on the queue and are redelivered. No offset injection or offset topic mirroring is required.
• Exactly-once delivery: The connector tracks progress in two places. It stores the sequence ID of the last committed batch in the Kafka Connect offset topic, and it holds the sequence ID and message IDs of in-flight messages in the MQ state queue that is set in mq.exactly.once.state.queue. Both must be preserved or migrated to maintain the exactly-once guarantee. For the full set of requirements, see exactly-once message delivery semantics.

Exactly-once delivery with the open-source connector also depends on settings that are not part of the Connector custom resource. Set exactly.once.source.support on the CFK Connect custom resource by using spec.configOverrides.server, run a single task, and retain the MQ state queue. For more information, see exactly-once delivery requirements.