Support for destination as key and reading MQMD

Author: AndrewJSchofield

config/mq-source.properties

@@ -1,4 +1,4 @@
-# Copyright 2017, 2018 IBM Corporation
+# Copyright 2017, 2018, 2019 IBM Corporation
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -56,7 +56,7 @@ mq.record.builder=com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBu
 # mq.message.body.jms=
 
 # The JMS message header to use as the Kafka record key - optional
-# Valid values are JMSMessageID, JMSCorrelationID and JMSCorrelationIDAsBytes
+# Valid values are JMSMessageID, JMSCorrelationID, JMSCorrelationIDAsBytes and JMSDestination
 # Don't forget to set key.converter to a compatible converter as described in README.md
 # mq.record.builder.key.header=
 
@@ -67,6 +67,9 @@ mq.record.builder=com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBu
 # The distinguished name pattern of the TLS (SSL) peer - optional
 # mq.ssl.peer.name=
 
+# Whether to enable reading of all MQMD fields (default false) - optional
+# mq.message.mqmd.read=
+
 # The name of the target Kafka topic - required
 topic=
 

pom.xml

@@ -20,7 +20,7 @@
   <groupId>com.ibm.eventstreams.connect</groupId>
   <artifactId>kafka-connect-mq-source</artifactId>
   <packaging>jar</packaging>
-  <version>1.0.2</version>
+  <version>1.1.0</version>
   <name>kafka-connect-mq-source</name>
     <organization>
     <name>IBM Corporation</name>

src/main/java/com/ibm/eventstreams/connect/mqsource/JMSReader.java

@@ -93,6 +93,7 @@ public void configure(Map<String, String> props) {
         String ccdtUrl = props.get(MQSourceConnector.CONFIG_NAME_MQ_CCDT_URL);
         String builderClass = props.get(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER);
         String mbj = props.get(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS);
+        String mdr = props.get(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_MQMD_READ);
         String sslCipherSuite = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_CIPHER_SUITE);
         String sslPeerName = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_PEER_NAME);
         String topic = props.get(MQSourceConnector.CONFIG_NAME_TOPIC);
@@ -157,6 +158,12 @@ else if (connectionMode.equals(MQSourceConnector.CONFIG_VALUE_MQ_CONNECTION_MODE
                 }
             }
 
+            if (mdr != null) {
+                if (Boolean.parseBoolean(mdr)) {
+                    queue.setBooleanProperty(WMQConstants.WMQ_MQMD_READ_ENABLED, true);
+                }
+            }
+
             this.topic = topic;
         }
         catch (JMSException | JMSRuntimeException jmse) {

src/main/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnector.java

@@ -83,6 +83,7 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSMESSAGEID = "JMSMessageID";
     public static final String CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONID = "JMSCorrelationID";
     public static final String CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONIDASBYTES = "JMSCorrelationIDAsBytes";
+    public static final String CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSDESTINATION = "JMSDestination";
 
     public static final String CONFIG_NAME_MQ_SSL_CIPHER_SUITE = "mq.ssl.cipher.suite"; 
     public static final String CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE = "The name of the cipher suite for the TLS (SSL) connection.";
@@ -98,11 +99,15 @@ public class MQSourceConnector extends SourceConnector {
     public static final int CONFIG_VALUE_MQ_BATCH_SIZE_DEFAULT = 250;
     public static final int CONFIG_VALUE_MQ_BATCH_SIZE_MINIMUM = 1;
 
+    public static final String CONFIG_NAME_MQ_MESSAGE_MQMD_READ = "mq.message.mqmd.read";
+    public static final String CONFIG_DOCUMENTATION_MQ_MESSAGE_MQMD_READ = "Whether to enable reading of all MQMD fields.";
+    public static final String CONFIG_DISPLAY_MQ_MESSAGE_MQMD_READ = "Enable reading of MQMD fields";
+
     public static final String CONFIG_NAME_TOPIC = "topic"; 
     public static final String CONFIG_DOCUMENTATION_TOPIC = "The name of the target Kafka topic.";
     public static final String CONFIG_DISPLAY_TOPIC = "Target Kafka topic";
 
-    public static String VERSION = "1.0.2";
+    public static String VERSION = "1.1.0";
 
     private Map<String, String> configProps;
 
@@ -240,7 +245,11 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_BATCH_SIZE, CONFIG_GROUP_MQ, 14, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_BATCH_SIZE);
 
-        config.define(CONFIG_NAME_TOPIC, Type.STRING, null, Importance.HIGH,
+        config.define(CONFIG_NAME_MQ_MESSAGE_MQMD_READ, Type.BOOLEAN, Boolean.FALSE, Importance.LOW,
+                      CONFIG_DOCUMENTATION_MQ_MESSAGE_MQMD_READ, CONFIG_GROUP_MQ, 15, Width.SHORT,
+                      CONFIG_DISPLAY_MQ_MESSAGE_MQMD_READ);
+                  
+        config.define(CONFIG_NAME_TOPIC, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
                       CONFIG_DOCUMENTATION_TOPIC, null, 0, Width.MEDIUM,
                       CONFIG_DISPLAY_TOPIC);
 

src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java

@@ -37,7 +37,7 @@
 public abstract class BaseRecordBuilder implements RecordBuilder {
     private static final Logger log = LoggerFactory.getLogger(BaseRecordBuilder.class);
 
-    public enum KeyHeader {NONE, MESSAGE_ID, CORRELATION_ID, CORRELATION_ID_AS_BYTES};
+    public enum KeyHeader {NONE, MESSAGE_ID, CORRELATION_ID, CORRELATION_ID_AS_BYTES, DESTINATION};
     protected KeyHeader keyheader = KeyHeader.NONE;
 
     /**
@@ -64,8 +64,11 @@ else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_J
                 keyheader = KeyHeader.CORRELATION_ID_AS_BYTES;
                 log.debug("Setting Kafka record key from JMSCorrelationIDAsBytes header field");
             }
-            else
-            {
+            else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSDESTINATION)) {
+                keyheader = KeyHeader.DESTINATION;
+                log.debug("Setting Kafka record key from JMSDestination header field");
+            }
+            else {
                 log.error("Unsupported MQ record builder key header value {}", kh);
                 throw new ConnectException("Unsupported MQ record builder key header value");
             }
@@ -115,6 +118,10 @@ public SchemaAndValue getKey(JMSContext context, String topic, Message message)
                 keySchema = Schema.OPTIONAL_BYTES_SCHEMA;
                 key = message.getJMSCorrelationIDAsBytes();
                 break;
+            case DESTINATION:
+                keySchema = Schema.OPTIONAL_STRING_SCHEMA;
+                key = message.getJMSDestination().toString();
+                break;
             default:
                 break;
         }

src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java

@@ -81,4 +81,4 @@ else if (message instanceof TextMessage) {
 
         return converter.toConnectData(topic, payload);
     }
-}
+}