Merge pull request #2 from mhub/feat/1670-authentication

Feat/1670 authentication

Author: katheris

README.md

@@ -155,13 +155,54 @@ The configuration options for the Kafka Connect source connector for IBM MQ are
 | 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.ccdt.url                  | The URL for the CCDT file containing MQ connection details  | string  |               | URL for obtaining a CCDT file                           |
 | 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                                              |
 
+### Using a CCDT file
+Some of the connection details for MQ can be provided in a [CCDT file](https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.0.0/com.ibm.mq.con.doc/q016730_.htm) by setting `mq.ccdt.url` in the Kafka Connect source connector configuration file. If using a CCDT file the `mq.connection.name.list` and `mq.channel.name` configuration options are not required.
+
+### Externalizing secrets
+[KIP 297](https://cwiki.apache.org/confluence/display/KAFKA/KIP-297%3A+Externalizing+Secrets+for+Connect+Configurations) introduced a mechanism to externalize secrets to be used as configuration for Kafka connectors.
+
+#### Example: externalizing secrets with FileConfigProvider
+
+Given a file `secrets.properties` with the contents:
+```
+secret-key=password
+```
+
+Update the worker configuration file to specify the FileConfigProvider which is included by default:
+
+```
+# Additional properties for the worker configuration to enable use of ConfigProviders
+# multiple comma-separated provider types can be specified here
+config.providers=file
+config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
+```
+
+Update the connector configuration file to reference `secret-key` in the file:
+
+```
+mq.password=${file:mq-secret.properties:secret-key}
+```
+
+#### Using custom Config Providers
+
+Custom config providers can also be enabled in the worker configuration file:
+```
+# Additional properties for the worker configuration to enable use of ConfigProviders
+# multiple comma-separated provider types can be specified here
+config.providers=file,other-provider
+config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
+# Other ConfigProvider implementations might require parameters passed in to configure() as follows:
+config.providers.other-provider.param.foo=value1
+config.providers.other-provider.param.bar=value2
+```
 
 ## Troubleshooting
 

config/mq-source.properties

@@ -21,10 +21,10 @@ tasks.max=1
 # The name of the MQ queue manager - required
 mq.queue.manager=
 
-# A list of one or more host(port) entries for connecting to the queue manager. Entries are separated with a comma - required
+# A list of one or more host(port) entries for connecting to the queue manager. Entries are separated with a comma - required (unless using CCDT)
 mq.connection.name.list=
 
-# The name of the server-connection channel - required
+# The name of the server-connection channel - required (unless using CCDT)
 mq.channel.name=
 
 # The name of the source MQ queue - required
@@ -36,6 +36,14 @@ mq.queue=
 # The password for authenticating with the queue manager - optional
 # mq.password=
 
+# Alternatively can use a ConfigProvider for externalising secrets (see README.md for more details)
+# Variable references are of the form ${provider:[path:]key} where the path is optional,
+# depending on the ConfigProvider implementation.
+# mq.password=${file:/var/run/secret.properties:secret-key}
+
+# The CCDT URL to use to establish a connection to the queue manager - optional
+# mq.ccdt.url=
+
 # The record builders control conversion of data between the messages in MQ and the internal Kafka Connect representation - required
 mq.record.builder=com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder
 # mq.record.builder=com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder

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

@@ -22,6 +22,8 @@
 import com.ibm.mq.jms.*;
 import com.ibm.msg.client.wmq.WMQConstants;
 
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -87,6 +89,7 @@ public void configure(Map<String, String> props) {
         String queueName = props.get(MQSourceConnector.CONFIG_NAME_MQ_QUEUE);
         String userName = props.get(MQSourceConnector.CONFIG_NAME_MQ_USER_NAME);
         String password = props.get(MQSourceConnector.CONFIG_NAME_MQ_PASSWORD);
+        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 sslCipherSuite = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_CIPHER_SUITE);
@@ -97,10 +100,22 @@ public void configure(Map<String, String> props) {
             mqConnFactory = new MQConnectionFactory();
             mqConnFactory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
             mqConnFactory.setQueueManager(queueManager);
-            mqConnFactory.setConnectionNameList(connectionNameList);
-            mqConnFactory.setChannel(channelName);
             mqConnFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
 
+            if (ccdtUrl != null) {
+                URL ccdtUrlObject;
+                try {
+                    ccdtUrlObject = new URL(ccdtUrl);
+                } catch (MalformedURLException e) {
+                    log.error("MalformedURLException exception {}", e);
+                    throw new ConnectException("CCDT file url invalid.");
+                }
+                mqConnFactory.setCCDTURL(ccdtUrlObject);
+            } else {
+                mqConnFactory.setConnectionNameList(connectionNameList);
+                mqConnFactory.setChannel(channelName);
+            }
+
             queue = new MQQueue(queueName);
 
             this.userName = userName;

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

@@ -59,6 +59,10 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_DOCUMENTATION_MQ_PASSWORD = "The password for authenticating with the queue manager.";
     public static final String CONFIG_DISPLAY_MQ_PASSWORD = "Password";
 
+    public static final String CONFIG_NAME_MQ_CCDT_URL = "mq.ccdt.url"; 
+    public static final String CONFIG_DOCUMENTATION_MQ_CCDT_URL = "The CCDT URL to use to establish a connection to the queue manager.";
+    public static final String CONFIG_DISPLAY_MQ_CCDT_URL = "CCDT URL";
+
     public static final String CONFIG_NAME_MQ_RECORD_BUILDER = "mq.record.builder";
     public static final String CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER = "The class used to build the Kafka Connect records.";
     public static final String CONFIG_DISPLAY_MQ_RECORD_BUILDER = "Record builder";
@@ -162,44 +166,48 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_QUEUE_MANAGER, CONFIG_GROUP_MQ, 1, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_QUEUE_MANAGER);
 
-        config.define(CONFIG_NAME_MQ_CONNECTION_NAME_LIST, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
+        config.define(CONFIG_NAME_MQ_CONNECTION_NAME_LIST, Type.STRING, null, Importance.MEDIUM,
                       CONFIG_DOCUMENTATION_MQ_CONNNECTION_NAME_LIST, CONFIG_GROUP_MQ, 2, Width.LONG,
                       CONFIG_DISPLAY_MQ_CONNECTION_NAME_LIST);
 
-        config.define(CONFIG_NAME_MQ_CHANNEL_NAME, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
+        config.define(CONFIG_NAME_MQ_CHANNEL_NAME, Type.STRING, null, Importance.MEDIUM,
                       CONFIG_DOCUMENTATION_MQ_CHANNEL_NAME, CONFIG_GROUP_MQ, 3, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_CHANNEL_NAME);
 
+        config.define(CONFIG_NAME_MQ_CCDT_URL, Type.STRING, null, Importance.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_CCDT_URL, CONFIG_GROUP_MQ, 4, Width.MEDIUM,
+                      CONFIG_DISPLAY_MQ_CCDT_URL);
+
         config.define(CONFIG_NAME_MQ_QUEUE, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
-                      CONFIG_DOCUMENTATION_MQ_QUEUE, CONFIG_GROUP_MQ, 4, Width.LONG,
+                      CONFIG_DOCUMENTATION_MQ_QUEUE, CONFIG_GROUP_MQ, 5, Width.LONG,
                       CONFIG_DISPLAY_MQ_QUEUE);
 
         config.define(CONFIG_NAME_MQ_USER_NAME, Type.STRING, null, Importance.MEDIUM,
-                      CONFIG_DOCUMENTATION_MQ_USER_NAME, CONFIG_GROUP_MQ, 5, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_USER_NAME, CONFIG_GROUP_MQ, 6, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_USER_NAME);
 
         config.define(CONFIG_NAME_MQ_PASSWORD, Type.PASSWORD, null, Importance.MEDIUM,
-                      CONFIG_DOCUMENTATION_MQ_PASSWORD, CONFIG_GROUP_MQ, 6, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_PASSWORD, CONFIG_GROUP_MQ, 7, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_PASSWORD);
 
         config.define(CONFIG_NAME_MQ_RECORD_BUILDER, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
-                      CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER, CONFIG_GROUP_MQ, 7, Width.LONG,
+                      CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER, CONFIG_GROUP_MQ, 8, Width.LONG,
                       CONFIG_DISPLAY_MQ_RECORD_BUILDER);
 
         config.define(CONFIG_NAME_MQ_MESSAGE_BODY_JMS, Type.BOOLEAN, Boolean.FALSE, Importance.MEDIUM,
-                      CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS, CONFIG_GROUP_MQ, 8, Width.SHORT,
+                      CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS, CONFIG_GROUP_MQ, 9, 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_DOCUMENTATION_MQ_RECORD_BUILDER_KEY_HEADER, CONFIG_GROUP_MQ, 10, 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, 10, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE, CONFIG_GROUP_MQ, 11, 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, 11, Width.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_PEER_NAME, CONFIG_GROUP_MQ, 12, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_SSL_PEER_NAME);
 
         config.define(CONFIG_NAME_TOPIC, Type.STRING, null, Importance.HIGH,

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

@@ -31,8 +31,8 @@ public class MQSourceTask extends SourceTask {
     private static final Logger log = LoggerFactory.getLogger(MQSourceTask.class);
 
     private static int BATCH_SIZE = 100;
-    private static int MAXUMSGS = 10000;
-    private static int MAXUMSGS_DELAY_MS = 500;
+    private static int MAX_UNCOMMITTED_MSGS = 10000;
+    private static int MAX_UNCOMMITTED_MSGS_DELAY_MS = 500;
 
     private JMSReader reader;
     private AtomicInteger uncommittedMessages = new AtomicInteger(0);
@@ -81,27 +81,27 @@ public MQSourceTask() {
 
         final List<SourceRecord> msgs = new ArrayList<>();
         int messageCount = 0;
-        int uncom = this.uncommittedMessages.get();
+        int uncommittedMessagesInt = this.uncommittedMessages.get();
 
-        if (uncom < MAXUMSGS) {
+        if (uncommittedMessagesInt < MAX_UNCOMMITTED_MSGS) {
             log.info("Polling for records");
 
             SourceRecord src;
             do {
                 // For the first message in the batch, wait a while if no message
-                src = reader.receive(messageCount == 0 ? true : false);
+                src = reader.receive(messageCount == 0);
                 if (src != null) {
                     msgs.add(src);
                     messageCount++;
-                    uncom = this.uncommittedMessages.incrementAndGet();
+                    uncommittedMessagesInt = this.uncommittedMessages.incrementAndGet();
                 }
-            } while ((src != null) && (messageCount < BATCH_SIZE) && (uncom < MAXUMSGS));
+            } while ((src != null) && (messageCount < BATCH_SIZE) && (uncommittedMessagesInt < MAX_UNCOMMITTED_MSGS));
 
             log.debug("Poll returning {} records", messageCount);
         }
         else {
             log.info("Uncommitted message limit reached");
-            Thread.sleep(MAXUMSGS_DELAY_MS);
+            Thread.sleep(MAX_UNCOMMITTED_MSGS_DELAY_MS);
         }
 
         log.trace("[{}]  Exit {}.poll, retval={}", Thread.currentThread().getId(), this.getClass().getName(), messageCount);