feat: Optional CCDT

Contributes to: mhub/qp-planning#1670

Signed-off-by: Andrew Dunnings <[email protected]>

Author: andrewdunnings

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 - equired (unless using CCDT)
 mq.channel.name=
 
 # The name of the source MQ queue - required
@@ -36,6 +36,9 @@ mq.queue=
 # The password for authenticating with the queue manager - optional
 # mq.password=
 
+# The CCDT URL to use to establish a connection to the queue manager - optional
+# mq.ccd.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,9 +100,21 @@ 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);
+                // mqConnFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
+            }
 
             queue = new MQQueue(queueName);
 

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";
@@ -182,6 +186,10 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_PASSWORD, CONFIG_GROUP_MQ, 6, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_PASSWORD);
 
+        config.define(CONFIG_NAME_MQ_CCDT_URL, Type.STRING, null, Importance.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_CCDT_URL, CONFIG_GROUP_MQ, 6, Width.MEDIUM,
+                      CONFIG_DISPLAY_MQ_CCDT_URL);
+
         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_DISPLAY_MQ_RECORD_BUILDER);

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

@@ -31,7 +31,7 @@ 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 = 10000; // What does this stand for, better name?
     private static int MAXUMSGS_DELAY_MS = 500;
 
     private JMSReader reader;
@@ -89,7 +89,7 @@ public MQSourceTask() {
             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++;