Merge pull request #114 from jhughes24816/fix-multiple-jms-connections

Fix multiple jms connections

Author: jhughes24816

.github/ISSUE_TEMPLATE/BUG-REPORT.yml

@@ -57,8 +57,8 @@ body:
       label: Version
       description: What version of our software are you running?
       options:
-        - 1.3.2 (Default)
-        - older (<1.3.2)
+        - 1.3.3 (Default)
+        - older (<1.3.3)
     validations:
       required: true
   - type: textarea

Dockerfile

@@ -10,7 +10,7 @@ COPY --chown=esuser:esgroup --from=builder /opt/kafka/libs/ /opt/kafka/libs/
 COPY --chown=esuser:esgroup --from=builder /opt/kafka/config/connect-distributed.properties /opt/kafka/config/
 COPY --chown=esuser:esgroup --from=builder /opt/kafka/config/connect-log4j.properties /opt/kafka/config/
 RUN mkdir /opt/kafka/logs && chown esuser:esgroup /opt/kafka/logs
-COPY --chown=esuser:esgroup target/kafka-connect-mq-source-1.3.2-jar-with-dependencies.jar /opt/kafka/libs/
+COPY --chown=esuser:esgroup target/kafka-connect-mq-source-1.3.3-jar-with-dependencies.jar /opt/kafka/libs/
 
 WORKDIR /opt/kafka
 

README.md

@@ -92,13 +92,13 @@ curl -X POST -H "Content-Type: application/json" http://localhost:8083/connector
 This repository includes an example Dockerfile to run Kafka Connect in distributed mode. It also adds in the MQ source connector as an available connector plugin. It uses the default `connect-distributed.properties` and `connect-log4j.properties` files.
 
 1. `mvn clean package`
-1. `docker build -t kafkaconnect-with-mq-source:1.3.2 .`
-1. `docker run -p 8083:8083 kafkaconnect-with-mq-source:1.3.2`
+1. `docker build -t kafkaconnect-with-mq-source:1.3.3 .`
+1. `docker run -p 8083:8083 kafkaconnect-with-mq-source:1.3.3`
 
 **NOTE:** To provide custom properties files create a folder called `config` containing the `connect-distributed.properties` and `connect-log4j.properties` files and use a Docker volume to make them available when running the container like this:
 
 ``` shell
-docker run -v $(pwd)/config:/opt/kafka/config -p 8083:8083 kafkaconnect-with-mq-source:1.3.2
+docker run -v $(pwd)/config:/opt/kafka/config -p 8083:8083 kafkaconnect-with-mq-source:1.3.3
 ```
 
 To start the MQ connector, you can use `config/mq-source.json` in this repository after replacing all placeholders and use a command like this:

pom.xml

@@ -20,7 +20,7 @@
   <groupId>com.ibm.eventstreams.connect</groupId>
   <artifactId>kafka-connect-mq-source</artifactId>
   <packaging>jar</packaging>
-  <version>1.3.2</version>
+  <version>1.3.3</version>
   <name>kafka-connect-mq-source</name>
     <organization>
     <name>IBM Corporation</name>
@@ -63,7 +63,7 @@
     <dependency>
       <groupId>com.ibm.mq</groupId>
       <artifactId>com.ibm.mq.allclient</artifactId>
-      <version>9.1.5.0</version>
+      <version>9.3.0.0</version>
     </dependency>
 
     <dependency>

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

@@ -79,7 +79,7 @@ public class JMSReader {
     private AtomicBoolean closeNow = new AtomicBoolean();           // Whether close has been requested
     private long reconnectDelayMillis = reconnectDelayMillisMin; // Delay between repeated reconnect attempts
 
-    private static long receiveTimeout = 30000L;
+    private static long receiveTimeout = 2000L;
     private static long reconnectDelayMillisMin = 64L;
     private static long reconnectDelayMillisMax = 8192L;
 
@@ -258,10 +258,8 @@ public SourceRecord receive(final boolean wait) {
         SourceRecord sr = null;
         try {
             if (wait) {
-                while (m == null && !closeNow.get()) {
-                    log.debug("Waiting {} ms for message", receiveTimeout);
-                    m = jmsCons.receive(receiveTimeout);
-                }
+                log.debug("Waiting {} ms for message", receiveTimeout);
+                m = jmsCons.receive(receiveTimeout);
 
                 if (m == null) {
                     log.debug("No message received");

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

@@ -136,7 +136,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 = "1.3.2";
+    public static String version = "1.3.3";
 
     private Map<String, String> configProps;
 

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

@@ -119,7 +119,7 @@ public List<SourceRecord> poll() throws InterruptedException {
 
         try {
             if (!stopNow.get()) {
-                log.info("Polling for records");
+                log.debug("Polling for records");
                 SourceRecord src;
                 do {
                     // For the first message in the batch, wait a while if no message