Enhance tracing and support MQ on Cloud

Author: AndrewJSchofield

.gitignore

@@ -3,8 +3,4 @@ target/
 .classpath
 .settings/
 .project
-.vscode/
-# Exclude local Maven repo used to incorporate downloaded MQ client JAR into the connector
-local-maven-repo/
-# Exclude MQ client JAR in the top level of the work tree
-allclient*.jar
+.vscode/

README.md

@@ -8,7 +8,7 @@ The connector is supplied as source code which you can easily build into a JAR f
 To build the connector, you must have the following installed:
 * [git](https://git-scm.com/)
 * [Maven](https://maven.apache.org)
-* Java 7 or later
+* Java 8 or later
 
 Clone the repository with the following command:
 ```shell
@@ -25,15 +25,15 @@ Build the connector using Maven:
 mvn clean package
 ```
 
-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.
+Once built, the output is a single JAR called `target/kafka-connect-mq-source-1.0-SNAPSHOT-jar-with-dependencies.jar` which contains all of the required dependencies.
 
 
 ## Running the connector
 To run the connector, you must have:
 * The JAR from building the connector
 * A properties file containing the configuration for the connector
-* Apache Kafka
-* IBM MQ v7.5 or later
+* Apache Kafka, either standalone or included as part of an offering such as IBM Event Streams
+* IBM MQ v8 or later, or the IBM MQ on Cloud service
 
 The connector can be run in a Kafka Connect worker in either standalone (single process) or distributed mode. It's a good idea to start in standalone mode.
 
@@ -139,6 +139,14 @@ You will need to put the public part of the client's certificate in the queue ma
 For troubleshooting, or to better understand the handshake performed by the IBM MQ Java client application in combination with your specific JSSE provider, you can enable debugging by setting `javax.net.debug=ssl` in the JVM environment.
 
 
+## Performance and syncpoint limit
+The connector uses a transacted JMS session to receive messages from MQ in syncpoint and periodically commits the in-flight transaction. This has the effect of batching messages together for improved efficiency. However, the frequency of committing transactions is controlled by the Kafka Connect framework rather than the connector. The connector is only able to receive up to the queue manager's maximum uncommitted message limit (typically 10000 messages) before committing.
+
+By default, Kafka Connect only commits every 60 seconds (10 seconds for the standalone worker), meaning that each task is limited to a rate of about 166 messages per second. You can increase the frequency of committing by using the `offset.flush.interval.ms` configuration in the worker configuration file. For example, if you set `offset.flush.interval.ms=5000`, the connector commits every 5 seconds increasing the maximum rate per task to about 2000 messages per second.
+
+If messages are being received faster than they can be committed, the connector prints a message `Uncommitted message limit reached` and sleeps for a short delay. You should use this as an indication to set the `offset.flush.interval.ms` to a lower value, or increase the number of tasks.
+
+
 ## Configuration
 The configuration options for the MQ Source Connector are as follows:
 
@@ -160,12 +168,14 @@ The configuration options for the MQ Source Connector are as follows:
 
 ## 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
 * JMX metrics
 * Separate TLS configuration for the connector so that keystore location and so on can be specified as configurations
 
 
+## Support
+A commercially supported version of this connector is available for customers with a support entitlement for [IBM Event Streams](https://developer.ibm.com/messaging/event-streams/).
+
+
 ## Issues and contributions
 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).
 

pom.xml

@@ -20,7 +20,7 @@
   <groupId>com.ibm.mq.kafkaconnect</groupId>
   <artifactId>kafka-connect-mq-source</artifactId>
   <packaging>jar</packaging>
-  <version>0.6-SNAPSHOT</version>
+  <version>1.0-SNAPSHOT</version>
   <name>kafka-connect-mq-source</name>
     <organization>
     <name>IBM Corporation</name>
@@ -45,13 +45,13 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>connect-api</artifactId>
-      <version>0.11.0.0</version>
+      <version>1.1.0</version>
       <scope>provided</scope>
       </dependency>
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>connect-json</artifactId>
-      <version>0.11.0.0</version>
+      <version>1.1.0</version>
       <scope>provided</scope>
     </dependency>
 

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

@@ -37,7 +37,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-
 /**
  * Reads messages from MQ using JMS. Uses a transacted session, adding messages to the current
  * transaction until told to commit. Automatically reconnects as needed.
@@ -76,6 +75,8 @@ public JMSReader() {}
      * @throws ConnectException   Operation failed and connector should stop.
      */
     public void configure(Map<String, String> props) {
+        log.trace("[{}] Entry {}.configure, props={}", Thread.currentThread().getId(), this.getClass().getName(), props);
+
         String queueManager = props.get(MQSourceConnector.CONFIG_NAME_MQ_QUEUE_MANAGER);
         String connectionNameList = props.get(MQSourceConnector.CONFIG_NAME_MQ_CONNECTION_NAME_LIST);
         String channelName = props.get(MQSourceConnector.CONFIG_NAME_MQ_CHANNEL_NAME);
@@ -94,6 +95,8 @@ public void configure(Map<String, String> props) {
             mqConnFactory.setQueueManager(queueManager);
             mqConnFactory.setConnectionNameList(connectionNameList);
             mqConnFactory.setChannel(channelName);
+            mqConnFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
+
             queue = new MQQueue(queueName);
 
             this.userName = userName;
@@ -119,7 +122,7 @@ public void configure(Map<String, String> props) {
             this.topic = topic;
         }
         catch (JMSException | JMSRuntimeException jmse) {
-            log.debug("JMS exception {}", jmse);
+            log.error("JMS exception {}", jmse);
             throw new ConnectException(jmse);
         }
 
@@ -129,15 +132,19 @@ public void configure(Map<String, String> props) {
             builder.configure(props);
         }
         catch (ClassNotFoundException | IllegalAccessException | InstantiationException | NullPointerException exc) {
-            log.debug("Could not instantiate message builder {}", builderClass);
+            log.error("Could not instantiate message builder {}", builderClass);
             throw new ConnectException("Could not instantiate message builder", exc);
         }
+
+        log.trace("[{}]  Exit {}.configure", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**
      * Connects to MQ.
      */
     public void connect() {
+        log.trace("[{}] Entry {}.connect", Thread.currentThread().getId(), this.getClass().getName());
+
         try {
             if (userName != null) {
                 jmsCtxt = mqConnFactory.createContext(userName, password, JMSContext.SESSION_TRANSACTED);
@@ -153,9 +160,11 @@ public void connect() {
         }
         catch (JMSRuntimeException jmse) {
             log.info("Connection to MQ could not be established");
-            log.debug("JMS exception {}", jmse);
+            log.error("JMS exception {}", jmse);
             handleException(jmse);
         }
+
+        log.trace("[{}]  Exit {}.connect", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**
@@ -166,7 +175,10 @@ public void connect() {
      * @return The SourceRecord representing the message
      */
     public SourceRecord receive(boolean wait) {
+        log.trace("[{}] Entry {}.receive", Thread.currentThread().getId(), this.getClass().getName());
+
         if (!connectInternal()) {
+            log.trace("[{}]  Exit {}.receive, retval=null", Thread.currentThread().getId(), this.getClass().getName());
             return null;
         }
 
@@ -176,12 +188,12 @@ public SourceRecord receive(boolean wait) {
             if (wait) {
                 while ((m == null) && !closeNow.get())
                 {
-                    log.trace("Waiting {} ms for message", RECEIVE_TIMEOUT);
+                    log.debug("Waiting {} ms for message", RECEIVE_TIMEOUT);
                     m = jmsCons.receive(RECEIVE_TIMEOUT);
                 }
 
                 if (m == null) {
-                    log.trace("No message received");
+                    log.debug("No message received");
                 }
             }
             else {
@@ -201,10 +213,11 @@ public SourceRecord receive(boolean wait) {
             }
         }
         catch (JMSException | JMSRuntimeException | ConnectException exc) {
-            log.debug("JMS exception {}", exc);
+            log.error("JMS exception {}", exc);
             handleException(exc);
         }
 
+        log.trace("[{}]  Exit {}.receive, retval={}", Thread.currentThread().getId(), this.getClass().getName(), sr);
         return sr;
     }
 
@@ -213,6 +226,8 @@ public SourceRecord receive(boolean wait) {
      * be processed, the transaction is "in peril" and is rolled back instead to avoid data loss.
      */
     public void commit() {
+        log.trace("[{}] Entry {}.commit", Thread.currentThread().getId(), this.getClass().getName());
+
         if (!connectInternal()) {
             return;
         }
@@ -223,7 +238,7 @@ public void commit() {
 
                 if (inperil) {
                     inperil = false;
-                    log.trace("Rolling back in-flight transaction");
+                    log.debug("Rolling back in-flight transaction");
                     jmsCtxt.rollback();
                 }
                 else {
@@ -232,15 +247,19 @@ public void commit() {
             }
         }
         catch (JMSRuntimeException jmse) {
-            log.debug("JMS exception {}", jmse);
+            log.error("JMS exception {}", jmse);
             handleException(jmse);
         }
+
+        log.trace("[{}]  Exit {}.commit", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**
      * Closes the connection.
      */
     public void close() {
+        log.trace("[{}] Entry {}.close", Thread.currentThread().getId(), this.getClass().getName());
+
         try {
             JMSContext ctxt = jmsCtxt;
             closeNow.set(true);
@@ -249,6 +268,8 @@ public void close() {
         catch (JMSRuntimeException jmse) {
             ;
         }
+
+        log.trace("[{}]  Exit {}.close", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**
@@ -262,9 +283,11 @@ private boolean connectInternal() {
         }
 
         if (closeNow.get()) {
+            log.debug("Closing connection now");
             return false;
         }
 
+        log.trace("[{}] Entry {}.connectInternal", Thread.currentThread().getId(), this.getClass().getName());
         try {
             if (userName != null) {
                 jmsCtxt = mqConnFactory.createContext(userName, password, JMSContext.SESSION_TRANSACTED);
@@ -279,18 +302,22 @@ private boolean connectInternal() {
             log.info("Connection to MQ established");
         }
         catch (JMSRuntimeException jmse) {
-            log.debug("JMS exception {}", jmse);
+            log.error("JMS exception {}", jmse);
             handleException(jmse);
+            log.trace("[{}]  Exit {}.connectInternal, retval=false", Thread.currentThread().getId(), this.getClass().getName());
             return false;
         }
 
+        log.trace("[{}]  Exit {}.connectInternal, retval=true", Thread.currentThread().getId(), this.getClass().getName());
         return true;
     }
 
     /**
      * Internal method to close the connection.
      */
     private void closeInternal() {
+        log.trace("[{}] Entry {}.closeInternal", Thread.currentThread().getId(), this.getClass().getName());
+
         try {
             inflight = false;
             inperil = false;
@@ -308,6 +335,8 @@ private void closeInternal() {
             jmsCtxt = null;
             log.debug("Connection to MQ closed");
         }
+
+        log.trace("[{}]  Exit {}.closeInternal", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**

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

@@ -86,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.6";
+    public static String VERSION = "1.0";
 
     private Map<String, String> configProps;
 
@@ -106,10 +106,14 @@ public class MQSourceConnector extends SourceConnector {
      * @param props configuration settings
      */
     @Override public void start(Map<String, String> props) {
+        log.trace("[{}] Entry {}.start, props={}", Thread.currentThread().getId(), this.getClass().getName(), props);
+
         configProps = props;
         for (final Entry<String, String> entry: props.entrySet()) {
-            log.trace("Connector props entry {} : {}", entry.getKey(), entry.getValue());
+            log.debug("Connector props entry {} : {}", entry.getKey(), entry.getValue());
         }
+
+        log.trace("[{}]  Exit {}.start", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**
@@ -127,19 +131,24 @@ public class MQSourceConnector extends SourceConnector {
      * @return configurations for Tasks
      */
     @Override public List<Map<String, String>> taskConfigs(int maxTasks) {
+        log.trace("[{}] Entry {}.taskConfigs, maxTasks={}", Thread.currentThread().getId(), this.getClass().getName(), maxTasks);
+
         List<Map<String, String>> taskConfigs = new ArrayList<>();
         for (int i = 0; i < maxTasks; i++)
         {
             taskConfigs.add(configProps);
         }
+
+        log.trace("[{}]  Exit {}.taskConfigs, retval={}", Thread.currentThread().getId(), this.getClass().getName(), taskConfigs);
         return taskConfigs;
     }
 
     /**
      * Stop this connector.
      */
     @Override public void stop() {
-
+        log.trace("[{}] Entry {}.stop", Thread.currentThread().getId(), this.getClass().getName());
+        log.trace("[{}]  Exit {}.stop", Thread.currentThread().getId(), this.getClass().getName());
     }
 
     /**