Migrating the IBM MQ sink connector

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

For general connector migration steps including how to stop the connector, ensure consumer lag is zero, and validate message delivery, see migrating connectors.

Migration paths

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

Note: Sink connectors track their position by using Kafka consumer groups rather than the Connect offset storage topic. Cluster Linking synchronizes consumer group offsets automatically regardless of which option you use. For more information, see sink connectors.

Option A: Confluent IBM MQ sink connector

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

Plugin packaging

The Confluent IBM MQ sink connector is available on Confluent Hub as confluentinc/kafka-connect-ibmmq-sink. It is a Confluent commercial connector that requires a Confluent Platform subscription after an evaluation period. For the licensing terms, the supported properties, and the prerequisites, see the connector overview and the configuration reference in the Confluent documentation.

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 install IBM MQ client library, 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 sink connector
   RUN confluent-hub install --no-prompt confluentinc/kafka-connect-ibmmq-sink:<version>
   
   # Copy the IBM MQ client library, which is required but not bundled in the Confluent Hub package.
   COPY com.ibm.mq.allclient.jar \
        /usr/share/confluent-hub-components/confluentinc-kafka-connect-ibmmq-sink/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-sink.

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 sink 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 sink 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-sink-connector
  namespace: <cp-namespace>
spec:
  class: io.confluent.connect.jms.IbmMqSinkConnector
  taskMax: 1
  configs:
    # Kafka source topic(s)
    topics: "MY.KAFKA.TOPIC"

    # 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"

    # Destination queue
    jms.destination.name: "KAFKA.SINK.QUEUE"
    jms.destination.type: "queue"

    # Message body format: string (default), json, avro, or bytes
    jms.message.format: "string"

    # Message delivery
    jms.producer.delivery.mode: "persistent"
    jms.producer.time.to.live.ms: "0"

    # Forward Kafka headers as JMS properties (optional)
    jms.forward.kafka.headers: "true"

    # Suppress RFH2 header for non-JMS MQ clients (optional)
    # mq.destination.suppress.rfh2: "true"

    # Exactly-once delivery (optional)
    # exactly.once.enabled: "true"
    # mq.offsets.queue.name: "KAFKA.OFFSETS"

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

  connectClusterRef:
    name: connect

Note: Before you set exactly.once.enabled to true, the queue that is specified in mq.offsets.queue.name must be created on the IBM MQ queue manager and dedicated to this connector. For the full set of requirements and limitations, see exactly once semantics in the Confluent documentation.

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 sink connector JAR files from your existing Event Streams Connect image, or download the open-source release from GitHub: ibm-messaging/kafka-connect-mq-sink.

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-sink-<version>-jar-with-dependencies.jar \
        /opt/kafka/plugins/mq-sink/
   

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

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 sink connector.

apiVersion: platform.confluent.io/v1beta1
kind: Connector
metadata:
  name: mq-sink-connector
  namespace: <cp-namespace>
spec:
  class: com.ibm.eventstreams.connect.mqsink.MQSinkConnector
  taskMax: 1
  configs:
    # Kafka source topic(s)
    topics: "MY.KAFKA.TOPIC"

    # MQ connection
    mq.queue.manager: "QM1"
    mq.connection.name.list: "<mq-host>(1414)"
    mq.channel.name: "KAFKA.CONN.SVRCONN"
    mq.queue: "KAFKA.SINK.QUEUE"
    mq.user.name: "app"
    mq.password: "<password>"
    mq.user.authentication.mqcsp: "true"

    # Message builder
    mq.message.builder: "com.ibm.eventstreams.connect.mqsink.builders.DefaultMessageBuilder"
    mq.message.body.jms: "true"

    # Message delivery
    mq.persistent: "true"
    mq.time.to.live: "0"
    mq.retry.backoff.ms: "60000"

    # MQMD and headers
    mq.message.mqmd.write: "true"
    mq.message.mqmd.context: "ALL"
    mq.message.builder.key.header: "JMSCorrelationID"
    mq.kafka.headers.copy.to.jms.properties: "true"

    # Reply-to queue (optional)
    mq.reply.queue: "KAFKA.REPLY.QUEUE"

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

  connectClusterRef:
    name: connect

Offset handling

Sink connectors track their read position in Kafka by using consumer groups, not the Connect offset storage topic.

If you are using Cluster Linking, ensure consumer.offset.sync.enable: "true" is set on the ClusterLink resource. Consumer group offsets are then synchronized automatically and no manual offset migration is required.

If you are using the IBM open-source MQ sink connector, offsets serve the following further purposes depending on your configuration:

• Message metadata enrichment: The connector can attach the source record’s Kafka offset to the outgoing JMS message as a long property by using mq.message.builder.offset.property. This property takes effect only when mq.message.body.jms is also set to true.
• Exactly-once delivery: In exactly-once mode, the connector records the last committed offset in the MQ state queue that is set in mq.exactly.once.state.queue, and uses it to skip records that were already delivered (record.kafkaOffset() <= lastCommittedOffset). This queue must be preserved or recreated on the target deployment. For the full set of requirements, see exactly-once message delivery semantics.
• Flush tracking: During flush(), the connector receives partition offset metadata to track its progress with the Kafka Connect framework. This requires no migration action.

Important: Before starting the new MQ sink connector, confirm that the consumer group has zero lag and that all messages processed by the Event Streams connector have been delivered to the MQ queue. For more information, see ensuring the consuming application has processed all messages.