--- title: Streaming XML messages from IBM MQ into Kafka into MongoDB published: true date: 2020-10-05 00:00:00 UTC tags: apachekafka,ibmmq,mongodb,dataintegration canonical_url: https://rmoff.net/2020/10/05/streaming-xml-messages-from-ibm-mq-into-kafka-into-mongodb/ --- Let’s imagine we have XML data on a queue in IBM MQ, and we want to ingest it into Kafka to then use downstream, perhaps in an application or maybe to stream to a NoSQL store like MongoDB. > **Note** > > This same pattern for ingesting XML will work with other connectors such as [JMS](https://www.confluent.io/hub/confluentinc/kafka-connect-jms) and [ActiveMQ](https://www.confluent.io/hub/confluentinc/kafka-connect-activemq). ![Streaming XML messages from IBM MQ through Apache Kafka into MongoDB](https://rmoff.net/images/2020/10/xml-ibmmq-mongodb.gif) I’ve got a [Docker Compose](https://github.com/confluentinc/demo-scene/blob/master/xml-to-kafka/docker-compose.yml) stack running that includes: - IBM MQ - Apache Kafka (deployed as Confluent Platform to include the all-important Schema Registry) - MongoDB Loading some test data onto IBM MQ ================================== Let’s load some messages onto the queue from an XML file: ```shell docker exec --interactive ibmmq \ /opt/mqm/samp/bin/amqsput DEV.QUEUE.1 QM1 < data/note.xml ``` ![](https://rmoff.net/images/2020/10/ibmmq.jpg) Streaming from IBM MQ to Kafka and translating the XML messages =============================================================== Now we can ingest this into Kafka using the [Kafka Connect](https://rmoff.net/2020/09/11/what-is-kafka-connect/) with the [IbmMQSourceConnector](https://www.confluent.io/hub/confluentinc/kafka-connect-ibmmq) plugin and [XML Transformation](https://www.confluent.io/hub/jcustenborder/kafka-connect-transform-xml): ```javascript curl -i -X PUT -H "Content-Type:application/json" \ http://localhost:8083/connectors/source-ibmmq-note-01/config \ -d '{ "connector.class": "io.confluent.connect.ibm.mq.IbmMQSourceConnector", "kafka.topic":"ibmmq-note-01", "mq.hostname":"ibmmq", "mq.port":"1414", "mq.queue.manager":"QM1", "mq.transport.type":"client", "mq.channel":"DEV.APP.SVRCONN", "mq.username":"app", "mq.password":"password123", "jms.destination.name":"DEV.QUEUE.1", "jms.destination.type":"queue", "confluent.license":"", "confluent.topic.bootstrap.servers":"broker:29092", "confluent.topic.replication.factor":"1", "transforms": "extractPayload,xml", "transforms.extractPayload.type": "org.apache.kafka.connect.transforms.ExtractField$Value", "transforms.extractPayload.field": "text", "transforms.xml.type": "com.github.jcustenborder.kafka.connect.transform.xml.FromXml$Value", "transforms.xml.schema.path": "file:///data/note.xsd", "value.converter": "io.confluent.connect.avro.AvroConverter", "value.converter.schema.registry.url":"http://schema-registry:8081" }' ``` > **Note** > > `ExtractField` is needed otherwise the XML transform will fail with `java.lang.UnsupportedOperationException: STRUCT is not a supported type.` since it will be trying to operate on the entire payload from IBM MQ which includes fields other than the XML that we’re interested in. The resulting Kafka topic holds the value of the `text` field in the messages, serialised in Avro: ```shell docker exec kafkacat \ kafkacat \ -b broker:29092 \ -r http://schema-registry:8081 \ -s key=s -s value=avro \ -t ibmmq-note-01 \ -C -o beginning -u -q -J | \ jq -c '.payload' ``` ```javascript {"Note":{"to":"Tove","from":"Jani","heading":"Reminder 01","body":"Don't forget me this weekend!"}} {"Note":{"to":"Jani","from":"Tove","heading":"Reminder 02","body":"Of course I won't!"}} … ``` To understand more about the concepts around getting XML data into Kafka [see here](https://dev.to/rmoff/ingesting-xml-data-into-kafka-introduction-2iha), and I’ve written about the specifics of Kafka Connect and the XML transformation [here](https://dev.to/rmoff/ingesting-xml-data-into-kafka-option-2-kafka-connect-plus-single-message-transform-5cml). Streaming the data from Kafka to MongoDB ======================================== We can then add another Kafka Connect connector to the pipeline, using the [official plugin for Kafka Connect from MongoDB](https://www.confluent.io/hub/mongodb/kafka-connect-mongodb), which will stream data straight from a Kafka topic into MongoDB: ```javascript curl -i -X PUT -H "Content-Type:application/json" \ http://localhost:8083/connectors/sink-mongodb-note-01/config \ -d '{ "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector", "topics":"ibmmq-note-01", "connection.uri":"mongodb://mongodb:27017", "database":"rmoff", "collection":"notes", "value.converter": "io.confluent.connect.avro.AvroConverter", "value.converter.schema.registry.url":"http://schema-registry:8081" }' ``` Check out the data in MongoDB: ```shell docker exec --interactive mongodb mongo localhost:27017 < Tove Jani Reminder 05 Srsly? " | docker exec --interactive ibmmq /opt/mqm/samp/bin/amqsput DEV.QUEUE.1 QM1 Sample AMQSPUT0 start target queue is DEV.QUEUE.1 Sample AMQSPUT0 end ``` And, behold, the new record in MongoDB: ```shell docker exec --interactive mongodb mongo localhost:27017 < Jani Tove Reminder 02 Of course I won't! ``` If we want to retain some or all of these fields, we’re going to have to approach things a different way. As things stand, there is no Single Message Transform that I’m aware of that can take *both* the non-XML fields *and* the XML field and wrangle them into a single structured schema (which is the ideal outcome, or perhaps putting the non-XML fields into the Kafka message header). By default the [IBM MQ Source Connector](https://docs.confluent.io/current/connect/kafka-connect-ibmmq/index.html) will write the full payload to a [schema](https://docs.confluent.io/current/connect/kafka-connect-ibmmq/index.html#io-confluent-connect-jms-value). This means that you still use a schema-supporting serialisation method, but the `text` payload field remains unparsed. Here’s an example: ```javascript curl -i -X PUT -H "Content-Type:application/json" \ http://localhost:8083/connectors/source-ibmmq-note-03/config \ -d '{ "connector.class": "io.confluent.connect.ibm.mq.IbmMQSourceConnector", "kafka.topic":"ibmmq-note-03", "mq.hostname":"ibmmq", "mq.port":"1414", "mq.queue.manager":"QM1", "mq.transport.type":"client", "mq.channel":"DEV.APP.SVRCONN", "mq.username":"app", "mq.password":"password123", "jms.destination.name":"DEV.QUEUE.1", "jms.destination.type":"queue", "confluent.license":"", "confluent.topic.bootstrap.servers":"broker:29092", "confluent.topic.replication.factor":"1", "value.converter": "io.confluent.connect.avro.AvroConverter", "value.converter.schema.registry.url":"http://schema-registry:8081" }' ``` Now the full IBM MQ message is written to a Kafka topic, serialised with a schema. We can deserialise it with something like kafkacat: ```shell kafkacat \ -b broker:29092 \ -r http://schema-registry:8081 \ -s key=s -s value=avro \ -t ibmmq-note-03 \ -C -c1 -o beginning -u -q -J | \ jq '.' ``` ```javascript { "topic": "ibmmq-note-03", "partition": 0, "offset": 0, "tstype": "create", "ts": 1601894073400, "broker": 1, "key": "Struct{messageID=ID:414d5120514d3120202020202020202060e67a5f033a2924}", "payload": { "messageID": "ID:414d5120514d3120202020202020202060e67a5f033a2924", "messageType": "text", "timestamp": 1601894073400, "deliveryMode": 1, "properties": { "JMS_IBM_Format": { "propertyType": "string", "boolean": null, "byte": null, "short": null, "integer": null, "long": null, "float": null, "double": null, "string": { "string": "MQSTR " } }, … "map": null, "text": { "string": " Tove Jani Reminder 01 Don't forget me this weekend! " } } } ``` Observe that the `text` field is just a string, holding \[what happens to be\] XML. You can use ksqlDB to work with the data, to an extent - although there’s currently no [support for handing the XML](https://github.com/confluentinc/ksql/issues/6357): ```sql SELECT "PROPERTIES"['JMSXAppID']->STRING as JMSXAppID, "PROPERTIES"['JMS_IBM_PutTime']->STRING as JMS_IBM_PutTime, "PROPERTIES"['JMSXDeliveryCount']->INTEGER as JMSXDeliveryCount, "PROPERTIES"['JMSXUserID']->STRING as JMSXUserID, text FROM IBMMQ_SOURCE EMIT CHANGES; ``` ```sql +-----------+-----------------+-------------------+------------+------------------------------------+ |JMSXAPPID |JMS_IBM_PUTTIME |JMSXDELIVERYCOUNT |JMSXUSERID |TEXT | +-----------+-----------------+-------------------+------------+------------------------------------+ |amqsput |10302905 |1 |mqm | Jani Tove Reminder 02 Of course I won't! | |amqsput |10302905 |1 |mqm | Tove Jani Reminder 03 Where are you? | ``` 👾 Try it out! ============= You can find the code to run this for yourself using Docker Compose on [GitHub](https://github.com/confluentinc/demo-scene/blob/master/xml-to-kafka/docker-compose.yml).

ibmmq

Streaming XML messages from IBM MQ into Kafka into MongoDB

Join Our Newsletter. No Spam, Only the good stuff.