feat: Expose one config adjust the blocking behavior

- Enhancement to the IBM MQ Source Connector to
  expose two configurable values to adjust the
  blocking behavior of the JMS receive methods when
  retrieving messages from MQ

Signed-off-by: Joel Hanson <[email protected]>

Author: Joel-hanson

src/integration/java/com/ibm/eventstreams/connect/mqsource/MQSourceTaskIT.java

@@ -666,7 +666,8 @@ public void testJmsWorkerWithCustomReciveForConsumerAndCustomReconnectValues() t
         connectorConfigProps.put("mq.message.body.jms", "true");
         connectorConfigProps.put("mq.record.builder",
                 "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
-        connectorConfigProps.put("mq.message.receive.timeout", "5000");
+        connectorConfigProps.put("mq.receive.timeout.ms", "5000");
+        connectorConfigProps.put("mq.receive.subsequent.timeout.ms", "2000");
         connectorConfigProps.put("mq.reconnect.delay.min.ms", "100");
         connectorConfigProps.put("mq.reconnect.delay.max.ms", "10000");
 
@@ -688,7 +689,8 @@ public void testJmsWorkerWithCustomReciveForConsumerAndCustomReconnectValues() t
         shared.attemptRollback();
         assertThat(stateMsgs1.size()).isEqualTo(1);
 
-        assertEquals(5000L, shared.getReceiveTimeout());
+        assertEquals(5000L, shared.getInitialReceiveTimeoutMs());
+        assertEquals(2000L, shared.getSubsequentReceiveTimeoutMs());
         assertEquals(100L, shared.getReconnectDelayMillisMin());
         assertEquals(10000L, shared.getReconnectDelayMillisMax());
     }

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

@@ -71,12 +71,17 @@ public class JMSWorker {
     private boolean connected = false; // Whether connected to MQ
     private AtomicBoolean closeNow; // Whether close has been requested
     private AbstractConfig config;
-    private long receiveTimeout; // Receive timeout for the jms consumer
+    private long initialReceiveTimeoutMs; // Receive timeout for the jms consumer
+    private long subsequentReceiveTimeoutMs; // Receive timeout for the jms consumer on the subsequent calls
     private long reconnectDelayMillisMin; // Delay between repeated reconnect attempts min
     private long reconnectDelayMillisMax; // Delay between repeated reconnect attempts max
 
-    long getReceiveTimeout() {
-        return receiveTimeout;
+    long getInitialReceiveTimeoutMs() {
+        return initialReceiveTimeoutMs;
+    }
+
+    long getSubsequentReceiveTimeoutMs() {
+        return subsequentReceiveTimeoutMs;
     }
 
     long getReconnectDelayMillisMin() {
@@ -144,7 +149,8 @@ public void configure(final AbstractConfig config) {
             userName = config.getString(MQSourceConnector.CONFIG_NAME_MQ_USER_NAME);
             password = config.getPassword(MQSourceConnector.CONFIG_NAME_MQ_PASSWORD);
             topic = config.getString(MQSourceConnector.CONFIG_NAME_TOPIC);
-            receiveTimeout = config.getLong(MQSourceConnector.CONFIG_MAX_RECEIVE_TIMEOUT);
+            initialReceiveTimeoutMs = config.getLong(MQSourceConnector.CONFIG_MAX_RECEIVE_TIMEOUT);
+            subsequentReceiveTimeoutMs = config.getLong(MQSourceConnector.CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT);
             reconnectDelayMillisMin = config.getLong(MQSourceConnector.CONFIG_RECONNECT_DELAY_MIN);
             reconnectDelayMillisMax = config.getLong(MQSourceConnector.CONFIG_RECONNECT_DELAY_MAX);
         } catch (JMSException | JMSRuntimeException jmse) {
@@ -222,10 +228,14 @@ public void connect() {
      *
      * @param queueName   The name of the queue to get messages from
      * @param queueConfig Any particular queue configuration that should be applied
-     * @param wait        Whether to wait indefinitely for a message
+     * @param initialCall Indicates whether this is the initial receive call in the polling cycle.
+     *                    Determines which configured timeout to use:
+     *                    - If true, uses the initial receive timeout.
+     *                    - If false, uses the subsequent receive timeout.
+     *                    A timeout value of 0 results in a non-blocking receiveNoWait() call.
      * @return The Message retrieved from MQ
      */
-    public Message receive(final String queueName, final QueueConfig queueConfig, final boolean wait) throws JMSRuntimeException, JMSException {
+    public Message receive(final String queueName, final QueueConfig queueConfig, final boolean initialCall) throws JMSRuntimeException, JMSException {
         log.trace("[{}] Entry {}.receive", Thread.currentThread().getId(), this.getClass().getName());
 
         if (!maybeReconnect()) {
@@ -243,21 +253,25 @@ public Message receive(final String queueName, final QueueConfig queueConfig, fi
             jmsConsumers.put(queueName, internalConsumer);
         }
 
-        Message message = null;
-        if (wait) {
-            log.debug("Waiting {} ms for message", receiveTimeout);
 
-            message = internalConsumer.receive(receiveTimeout);
+        final long timeoutMs = initialCall
+            ? initialReceiveTimeoutMs
+            : subsequentReceiveTimeoutMs;
+
+        Message message = null;
+        if (timeoutMs > 0) {
+            // block up to timeoutMs
+            message = internalConsumer.receive(timeoutMs);
 
             if (message == null) {
-                log.debug("No message received");
+                log.debug("No message received within {} ms on queue={}", timeoutMs, queueName);
             }
         } else {
+            // non‐blocking
             message = internalConsumer.receiveNoWait();
         }
 
         log.trace("[{}]  Exit {}.receive, retval={}", Thread.currentThread().getId(), this.getClass().getName(), message);
-
         return message;
     }
 

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

@@ -163,12 +163,18 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_VALUE_MQ_CLIENT_RECONNECT_OPTION_DISABLED = "DISABLED";
     public static final String CONFIG_VALUE_MQ_CLIENT_RECONNECT_OPTION_ASDEF = "ASDEF";
 
-    public static final String CONFIG_MAX_RECEIVE_TIMEOUT = "mq.message.receive.timeout";
+    public static final String CONFIG_MAX_RECEIVE_TIMEOUT = "mq.receive.timeout.ms";
     public static final String CONFIG_DOCUMENTATION_MAX_RECEIVE_TIMEOUT = "How long the connector should wait (in milliseconds) for a message to arrive if no message is available immediately";
-    public static final String CONFIG_DISPLAY_MAX_RECEIVE_TIMEOUT = "message receive timeout";
+    public static final String CONFIG_DISPLAY_MAX_RECEIVE_TIMEOUT = "Initial receive timeout (ms)";
     public static final long CONFIG_MAX_RECEIVE_TIMEOUT_DEFAULT = 2000L;
     public static final long CONFIG_MAX_RECEIVE_TIMEOUT_MINIMUM = 1L;
 
+    public static final String CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT = "mq.receive.subsequent.timeout.ms";
+    public static final String CONFIG_DOCUMENTATION_SUBSEQUENT_RECEIVE_TIMEOUT = "How long (in milliseconds) the connector should wait for subsequent receives, "
+            + "defaults to 0 (no-wait) and uses receiveNoWait().";
+    public static final String CONFIG_DISPLAY_SUBSEQUENT_RECEIVE_TIMEOUT = "subsequent receive timeout (ms)";
+    public static final long CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT_DEFAULT = 0L;
+
     public static final String CONFIG_RECONNECT_DELAY_MIN = "mq.reconnect.delay.min.ms";
     public static final String CONFIG_DOCUMENTATION_RECONNECT_DELAY_MIN = "The minimum delay in milliseconds for reconnect attempts.";
     public static final String CONFIG_DISPLAY_RECONNECT_DELAY_MIN = "reconnect minimum delay";
@@ -576,27 +582,38 @@ null, new ReadableFile(),
                 Width.SHORT,
                 CONFIG_DISPLAY_MQ_CLIENT_RECONNECT_OPTIONS);
         CONFIGDEF.define(CONFIG_MAX_RECEIVE_TIMEOUT,
-                Type.LONG,
-                CONFIG_MAX_RECEIVE_TIMEOUT_DEFAULT, ConfigDef.Range.atLeast(CONFIG_MAX_RECEIVE_TIMEOUT_MINIMUM),
-                Importance.MEDIUM,
+                ConfigDef.Type.LONG,
+                CONFIG_MAX_RECEIVE_TIMEOUT_DEFAULT,
+                ConfigDef.Range.atLeast(CONFIG_MAX_RECEIVE_TIMEOUT_MINIMUM),
+                ConfigDef.Importance.MEDIUM,
                 CONFIG_DOCUMENTATION_MAX_RECEIVE_TIMEOUT,
-                CONFIG_GROUP_MQ, 26,
-                Width.MEDIUM,
+                "MQ",
+                26,
+                ConfigDef.Width.MEDIUM,
                 CONFIG_DISPLAY_MAX_RECEIVE_TIMEOUT);
+        CONFIGDEF.define(CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT,
+                ConfigDef.Type.LONG,
+                CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT_DEFAULT,
+                ConfigDef.Importance.LOW,
+                CONFIG_DOCUMENTATION_SUBSEQUENT_RECEIVE_TIMEOUT,
+                "MQ",
+                27,
+                ConfigDef.Width.MEDIUM,
+                CONFIG_DISPLAY_SUBSEQUENT_RECEIVE_TIMEOUT);
         CONFIGDEF.define(CONFIG_RECONNECT_DELAY_MIN,
                 Type.LONG,
                 CONFIG_RECONNECT_DELAY_MIN_DEFAULT, ConfigDef.Range.atLeast(CONFIG_RECONNECT_DELAY_MIN_MINIMUM),
                 Importance.MEDIUM,
                 CONFIG_DOCUMENTATION_RECONNECT_DELAY_MIN,
-                CONFIG_GROUP_MQ, 27,
+                CONFIG_GROUP_MQ, 28,
                 Width.MEDIUM,
                 CONFIG_DISPLAY_RECONNECT_DELAY_MIN);
         CONFIGDEF.define(CONFIG_RECONNECT_DELAY_MAX,
                 Type.LONG,
                 CONFIG_RECONNECT_DELAY_MAX_DEFAULT, ConfigDef.Range.atLeast(CONFIG_RECONNECT_DELAY_MAX_MINIMUM),
                 Importance.MEDIUM,
                 CONFIG_DOCUMENTATION_RECONNECT_DELAY_MAX,
-                CONFIG_GROUP_MQ, 28,
+                CONFIG_GROUP_MQ, 29,
                 Width.MEDIUM,
                 CONFIG_DISPLAY_RECONNECT_DELAY_MAX);