Add support for Kafka keys

Author: AndrewJSchofield

README.md

@@ -25,7 +25,7 @@ Build the connector using Maven:
 mvn clean package
 ```
 
-Once built, the output is a single JAR called `target/kafka-connect-mq-source-0.5-SNAPSHOT-jar-with-dependencies.jar` which contains all of the required dependencies.
+Once built, the output is a single JAR called `target/kafka-connect-mq-source-0.6-SNAPSHOT-jar-with-dependencies.jar` which contains all of the required dependencies.
 
 
 ## Running the connector
@@ -91,7 +91,6 @@ mq.message.body.jms=true
 value.converter=org.apache.kafka.connect.storage.StringConverter
 ```
 
-
 ### The gory detail
 The messages received from MQ are processed by a record builder which builds a Kafka Connect record to represent the message. There are two record builders supplied with the MQ source connector. The connector has a configuration option *mq.message.body.jms* that controls whether it interprets the MQ messages as JMS messages or regular MQ messages.
 
@@ -113,6 +112,17 @@ You must then choose a converter than can handle the value schema and class. The
 
 In addition, there is another converter for the Avro format that is part of the Confluent Platform. This has not been tested with the MQ source connector at this time.
 
+### Key support and partitioning
+By default, the connector does not use keys for the Kafka messages it publishes. It can be configured to use the JMS message headers to set the key of the Kafka records. You could use this, for example, to use the MQMD correlation identifier as the partitioning key when the messages are published to Kafka. There are three valid values for the `mq.record.builder.key.header` that controls this behavior.
+
+| mq.record.builder.key.header | Key schema      | Key class | Recommended value for key.converter                    |
+| ---------------------------- |---------------- | --------- | ------------------------------------------------------ |
+| JMSMessageID                 | OPTIONAL_STRING | String    | org.apache.kafka.connect.storage.StringConverter       |
+| JMSCorrelationID             | OPTIONAL_STRING | String    | org.apache.kafka.connect.storage.StringConverter       |
+| JMSCorrelationIDAsBytes      | OPTIONAL_BYTES  | byte[]    | org.apache.kafka.connect.converters.ByteArrayConverter |
+
+In MQ, the message ID and correlation ID are both 24-byte arrays. As strings, the connector represents them using a sequence of 48 hexadecimal characters.
+
 
 ## Security
 The connector supports authentication with user name and password and also connections secured with TLS using a server-side certificate and mutual authentication with client-side certificates.
@@ -132,32 +142,32 @@ For troubleshooting, or to better understand the handshake performed by the IBM
 ## Configuration
 The configuration options for the MQ Source Connector are as follows:
 
-| Name                    | Description                                                 | Type    | Default       | Valid values                     |
-| ----------------------- | ----------------------------------------------------------- | ------- | ------------- | -------------------------------- |
-| mq.queue.manager        | The name of the MQ queue manager                            | string  |               | MQ queue manager name            |
-| mq.connection.name.list | List of connection names for queue manager                  | string  |               | host(port)[,host(port),...]      |
-| mq.channel.name         | The name of the server-connection channel                   | string  |               | MQ channel name                  |
-| mq.queue                | The name of the source MQ queue                             | string  |               | MQ queue name                    |
-| mq.user.name            | The user name for authenticating with the queue manager     | string  |               | User name                        |
-| mq.password             | The password for authenticating with the queue manager      | string  |               | Password                         |
-| mq.record.builder       | The class used to build the Kafka Connect record            | string  |               | Class implementing RecordBuilder |
-| mq.message.body.jms     | Whether to interpret the message body as a JMS message type | boolean | false         |                                  |
-| mq.ssl.cipher.suite     | The name of the cipher suite for TLS (SSL) connection       | string  |               | Blank or valid cipher suite      |
-| mq.ssl.peer.name        | The distinguished name pattern of the TLS (SSL) peer        | string  |               | Blank or DN pattern              |
-| topic                   | The name of the target Kafka topic                          | string  |               | Topic name                       |
+| Name                         | Description                                                 | Type    | Default       | Valid values                                            |
+| ---------------------------- | ----------------------------------------------------------- | ------- | ------------- | ------------------------------------------------------- |
+| mq.queue.manager             | The name of the MQ queue manager                            | string  |               | MQ queue manager name                                   |
+| mq.connection.name.list      | List of connection names for queue manager                  | string  |               | host(port)[,host(port),...]                             |
+| mq.channel.name              | The name of the server-connection channel                   | string  |               | MQ channel name                                         |
+| mq.queue                     | The name of the source MQ queue                             | string  |               | MQ queue name                                           |
+| mq.user.name                 | The user name for authenticating with the queue manager     | string  |               | User name                                               |
+| mq.password                  | The password for authenticating with the queue manager      | string  |               | Password                                                |
+| mq.record.builder            | The class used to build the Kafka Connect record            | string  |               | Class implementing RecordBuilder                        |
+| mq.message.body.jms          | Whether to interpret the message body as a JMS message type | boolean | false         |                                                         |
+| mq.record.builder.key.header | The JMS message header to use as the Kafka record key       | string  |               | JMSMessageID, JMSCorrelationID, JMSCorrelationIDAsBytes |
+| mq.ssl.cipher.suite          | The name of the cipher suite for TLS (SSL) connection       | string  |               | Blank or valid cipher suite                             |
+| mq.ssl.peer.name             | The distinguished name pattern of the TLS (SSL) peer        | string  |               | Blank or DN pattern                                     |
+| topic                        | The name of the target Kafka topic                          | string  |               | Topic name                                              |
 
 
 ## Future enhancements
 The connector is intentionally basic. The idea is to enhance it over time with additional features to make it more capable. Some possible future enhancements are:
 * Configurable schema for MQ messages
 * Simplification of handling message formats
-* Message key support
 * JMX metrics
 * Separate TLS configuration for the connector so that keystore location and so on can be specified as configurations
 
 
 ## Issues and contributions
-For issues relating specifically to this connect, please use the [GitHub issue tracker](https://github.com/ibm-messaging/kafka-connect-mq-source/issues). If you do submit a Pull Request related to this connector, please indicate in the Pull Request that you accept and agree to be bound by the terms of the [IBM Contributor License Agreement](CLA.md).
+For issues relating specifically to this connector, please use the [GitHub issue tracker](https://github.com/ibm-messaging/kafka-connect-mq-source/issues). If you do submit a Pull Request related to this connector, please indicate in the Pull Request that you accept and agree to be bound by the terms of the [IBM Contributor License Agreement](CLA.md).
 
 
 ## License

UsingMQwithKafkaConnect.md

@@ -154,7 +154,7 @@ topic=TSOURCE
 
 Change directory to the Kafka root directory. Start the connector worker replacing `<connector-root-directory>`:
 ``` shell
-CLASSPATH=<connector-root-directory>/target/kafka-connect-mq-source-0.5-SNAPSHOT-jar-with-dependencies.jar bin/connect-standalone.sh config/connect-standalone.properties ~/mq-source.properties
+CLASSPATH=<connector-root-directory>/target/kafka-connect-mq-source-0.6-SNAPSHOT-jar-with-dependencies.jar bin/connect-standalone.sh config/connect-standalone.properties ~/mq-source.properties
 ```
 
 Wait while the worker starts and then prints:
@@ -212,7 +212,7 @@ mq.password=passw0rd
 
 Change directory to the Kafka root directory. Start the connector worker replacing `<connector-root-directory>`:
 ``` shell
-CLASSPATH=<connector-root-directory>/target/kafka-connect-mq-sink-0.5-SNAPSHOT-jar-with-dependencies.jar bin/kafka-connect-standalone config/connect-standalone.properties ~/mq-sink.properties
+CLASSPATH=<connector-root-directory>/target/kafka-connect-mq-sink-0.6-SNAPSHOT-jar-with-dependencies.jar bin/kafka-connect-standalone config/connect-standalone.properties ~/mq-sink.properties
 ```
 
 Wait while the worker starts and then prints:

config/mq-source.properties

@@ -1,4 +1,4 @@
-# Copyright 2017 IBM Corporation
+# Copyright 2017, 2018 IBM Corporation
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -43,6 +43,11 @@ mq.record.builder=com.ibm.mq.kafkaconnect.builders.DefaultRecordBuilder
 # Whether to interpret the message body as a JMS message type (default false) - optional
 # mq.message.body.jms=
 
+# The JMS message header to use as the Kafka record key - optional
+# Valid values are JMSMessageID, JMSCorrelationID and JMSCorrelationIDAsBytes
+# Don't forget to set key.converter to a compatible converter as described in README.md
+# mq.record.builder.key.header=
+
 # The name of the cipher suite for TLS (SSL) connection (default blank, meaning do not use TLS) - optional
 # See https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.0.0/com.ibm.mq.dev.doc/q113220_.htm for valid values
 # mq.ssl.cipher.suite=
@@ -54,7 +59,6 @@ mq.record.builder=com.ibm.mq.kafkaconnect.builders.DefaultRecordBuilder
 topic=
 
 # The converters control conversion of data between the internal Kafka Connect representation and the messages in Kafka.
-# Note that ByteArrayConverter is new in Kafka 0.11.
 # key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
 # key.converter=org.apache.kafka.connect.storage.StringConverter
 # key.converter=org.apache.kafka.connect.json.JsonConverter

pom.xml

@@ -20,7 +20,7 @@
   <groupId>com.ibm.mq.kafkaconnect</groupId>
   <artifactId>kafka-connect-mq-source</artifactId>
   <packaging>jar</packaging>
-  <version>0.5-SNAPSHOT</version>
+  <version>0.6-SNAPSHOT</version>
   <name>kafka-connect-mq-source</name>
     <organization>
     <name>IBM Corporation</name>

src/main/java/com/ibm/mq/kafkaconnect/JMSReader.java

@@ -126,6 +126,7 @@ public void configure(Map<String, String> props) {
         try {
             Class<? extends RecordBuilder> c = Class.forName(builderClass).asSubclass(RecordBuilder.class);
             builder = c.newInstance();
+            builder.configure(props);
         }
         catch (ClassNotFoundException | IllegalAccessException | InstantiationException | NullPointerException exc) {
             log.debug("Could not instantiate message builder {}", builderClass);

src/main/java/com/ibm/mq/kafkaconnect/MQSourceConnector.java

@@ -67,6 +67,13 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS = "Whether to interpret the message body as a JMS message type.";
     public static final String CONFIG_DISPLAY_MQ_MESSAGE_BODY_JMS = "Message body as JMS";
 
+    public static final String CONFIG_NAME_MQ_RECORD_BUILDER_KEY_HEADER = "mq.record.builder.key.header";
+    public static final String CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_KEY_HEADER = "The JMS message header to use as the Kafka record key.";
+    public static final String CONFIG_DISPLAY_MQ_RECORD_BUILDER_KEY_HEADER = "Record builder key header";
+    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_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.";
     public static final String CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE = "SSL cipher suite";
@@ -79,7 +86,7 @@ public class MQSourceConnector extends SourceConnector {
     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 = "0.5";
+    public static String VERSION = "0.6";
 
     private Map<String, String> configProps;
 
@@ -174,12 +181,16 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS, CONFIG_GROUP_MQ, 8, Width.SHORT,
                       CONFIG_DISPLAY_MQ_MESSAGE_BODY_JMS);
 
+        config.define(CONFIG_NAME_MQ_RECORD_BUILDER_KEY_HEADER, Type.STRING, null, Importance.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_KEY_HEADER, CONFIG_GROUP_MQ, 9, Width.MEDIUM,
+                      CONFIG_DISPLAY_MQ_RECORD_BUILDER_KEY_HEADER);
+
         config.define(CONFIG_NAME_MQ_SSL_CIPHER_SUITE, Type.STRING, null, Importance.MEDIUM,
-                      CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE, CONFIG_GROUP_MQ, 9, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE, CONFIG_GROUP_MQ, 10, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE);
 
         config.define(CONFIG_NAME_MQ_SSL_PEER_NAME, Type.STRING, null, Importance.MEDIUM,
-                      CONFIG_DOCUMENTATION_MQ_SSL_PEER_NAME, CONFIG_GROUP_MQ, 10, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_PEER_NAME, CONFIG_GROUP_MQ, 11, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_SSL_PEER_NAME);
 
         config.define(CONFIG_NAME_TOPIC, Type.STRING, null, Importance.HIGH,