feat: Config changes and new unit tests

Signed-off-by: Joel Hanson <joelhanson025@gmail.com>

Author: Joel-hanson

README.md

@@ -308,6 +308,7 @@ The configuration options for the Kafka Connect source connector for IBM MQ are
 | `mq.message.receive.timeout`                    | The timeout (in milliseconds) for receiving messages from the queue manager before returning to Kafka Connect.                                                          | long    | 2000           | 1 or greater                                                                                                                                                                        |
 | `mq.reconnect.delay.min.ms`                     | The minimum delay (in milliseconds) for reconnecting to the queue manager after a connection error.                                                                     | long    | 64             | 1 or greater                                                                                                                                                                        |
 | `mq.reconnect.delay.max.ms`                     | The maximum delay (in milliseconds) for reconnecting to the queue manager after a connection error.                                                                     | long    | 8192           | 1 or greater                                                                                                                                                                        |
+| `mq.receive.max.poll.time.ms`                   | Maximum time (in milliseconds) to poll messages in a single Kafka Connect task cycle. If set to 0, polling continues until batch size or a receive returns null.        | long    | 0              | 0 or greater                                                                                                                                                                        |
 | `errors.deadletterqueue.topic.name`             | The name of the Kafka topic to use as the dead letter queue (DLQ) for poison messages that fail during processing within the record builder component of the connector. | string  |                | If left blank (default), failed messages will not be written to a DLQ.                                                                                                              |
 | `errors.deadletterqueue.context.headers.enable` | Whether to add error context headers to messages written to the DLQ.                                                                                                    | boolean | false          | When enabled, additional headers describing the error will be included with each DLQ record.                                                                                        |
 

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

@@ -1300,7 +1300,7 @@ public void verifyLoggingErrorsWithMessageHavingDefaultRecordBuilder() throws Ex
         assertThat(processedRecords.get(1).topic()).isEqualTo("mytopic");
     }
 
-    public void testMaxPollTimeTerminatesBatchEarly() throws Exception {
+    public void testMaxPollTimeTerminates() throws Exception {
         connectTask = getSourceTaskWithEmptyKafkaOffset();
 
         final Map<String, String> connectorConfigProps = createExactlyOnceConnectorProperties();
@@ -1309,7 +1309,7 @@ public void testMaxPollTimeTerminatesBatchEarly() throws Exception {
                 "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
         connectorConfigProps.put("mq.message.receive.timeout", "150");
         connectorConfigProps.put("mq.receive.subsequent.timeout.ms", "10");
-        connectorConfigProps.put("mq.receive.max.poll.time.ms", "200"); // stop after 200ms
+        connectorConfigProps.put("mq.receive.max.poll.time.ms", "200");
         connectorConfigProps.put("mq.batch.size", "5000");
 
         final JMSWorker shared = new JMSWorker();
@@ -1320,25 +1320,131 @@ public void testMaxPollTimeTerminatesBatchEarly() throws Exception {
 
         connectTask.start(connectorConfigProps, shared, dedicated, sequenceStateClient);
 
-        List<Message> messages = createAListOfMessages(getJmsContext(), 10, "msg ");
+        final List<Message> messages = createAListOfMessages(getJmsContext(), 10, "msg ");
         putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, messages);
-        messages = createAListOfMessages(getJmsContext(), 10, "msg ");
         putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, messages);
-        messages = createAListOfMessages(getJmsContext(), 10, "msg ");
         putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, messages);
 
+        final List<SourceRecord> kafkaMessages = connectTask.poll();
+
+        final List<Message> stateMsgs1 = browseAllMessagesFromQueue(DEFAULT_STATE_QUEUE);
+        assertThat(stateMsgs1.size()).isEqualTo(1);
+        final List<Message> sourceMsgs = getAllMessagesFromQueue(DEFAULT_SOURCE_QUEUE);
+        assertThat(sourceMsgs.size()).isEqualTo(0);
+        assertEquals(30, kafkaMessages.size());
+    }
+
+    @Test
+    public void testMaxPollTimeTerminatesBatchEarly() throws Exception {
+        connectTask = getSourceTaskWithEmptyKafkaOffset();
+
+        final Map<String, String> connectorConfigProps = createExactlyOnceConnectorProperties();
+        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", "100");
+        connectorConfigProps.put("mq.receive.subsequent.timeout.ms", "10");
+        connectorConfigProps.put("mq.receive.max.poll.time.ms", "200"); // stop after 200ms
+        connectorConfigProps.put("mq.batch.size", "5000");
+
+        final JMSWorker shared = new JMSWorker();
+        shared.configure(getPropertiesConfig(connectorConfigProps));
+        final JMSWorker dedicated = new JMSWorker();
+        dedicated.configure(getPropertiesConfig(connectorConfigProps));
+        final SequenceStateClient sequenceStateClient = new SequenceStateClient(DEFAULT_STATE_QUEUE, shared, dedicated);
+
+        connectTask.start(connectorConfigProps, shared, dedicated, sequenceStateClient);
+
+        final List<Message> messages = createAListOfMessages(getJmsContext(), 10, "msg ");
+        putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, messages);
         final long start = System.nanoTime();
         final List<SourceRecord> kafkaMessages = connectTask.poll();
         final long durationMs = (System.nanoTime() - start) / 1_000_000;
 
-        System.out.println(durationMs);
         // Poll should end close to 200ms
-        assertThat(durationMs >= 180 && durationMs <= 500).isTrue();
+        assertThat(durationMs <= 210).isTrue();
 
         final List<Message> stateMsgs1 = browseAllMessagesFromQueue(DEFAULT_STATE_QUEUE);
         assertThat(stateMsgs1.size()).isEqualTo(1);
         final List<Message> sourceMsgs = getAllMessagesFromQueue(DEFAULT_SOURCE_QUEUE);
         assertThat(sourceMsgs.size()).isEqualTo(0);
-        assertEquals(30, kafkaMessages.size());
+        assertEquals(10, kafkaMessages.size());
+    }
+
+    @Test
+    public void testPollEndsWhenBatchSizeReached() throws Exception {
+        connectTask = getSourceTaskWithEmptyKafkaOffset();
+
+        final Map<String, String> config = createExactlyOnceConnectorProperties();
+        config.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
+        config.put("mq.message.receive.timeout", "100");
+        config.put("mq.receive.subsequent.timeout.ms", "10");
+        config.put("mq.receive.max.poll.time.ms", "1000");
+        config.put("mq.batch.size", "10");
+
+        final JMSWorker shared = new JMSWorker();
+        shared.configure(getPropertiesConfig(config));
+        final JMSWorker dedicated = new JMSWorker();
+        dedicated.configure(getPropertiesConfig(config));
+        final SequenceStateClient sequenceStateClient = new SequenceStateClient(DEFAULT_STATE_QUEUE, shared, dedicated);
+        connectTask.start(config, shared, dedicated, sequenceStateClient);
+
+        putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, createAListOfMessages(getJmsContext(), 12, "msg "));
+
+        final long start = System.nanoTime();
+        connectTask.poll();
+        final long durationMs = (System.nanoTime() - start) / 1_000_000;
+
+        assertThat(durationMs < 1000).isTrue();
+    }
+
+    @Test
+    public void testPollWithMaxPollTimeZeroBehavesAsDefault() throws Exception {
+        connectTask = getSourceTaskWithEmptyKafkaOffset();
+        final Map<String, String> config = createExactlyOnceConnectorProperties();
+        config.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
+        config.put("mq.message.receive.timeout", "400");
+        config.put("mq.receive.max.poll.time.ms", "0");
+        config.put("mq.batch.size", "100");
+
+        final JMSWorker shared = new JMSWorker();
+        shared.configure(getPropertiesConfig(config));
+        final JMSWorker dedicated = new JMSWorker();
+        dedicated.configure(getPropertiesConfig(config));
+        final SequenceStateClient sequenceStateClient = new SequenceStateClient(DEFAULT_STATE_QUEUE, shared, dedicated);
+        connectTask.start(config, shared, dedicated, sequenceStateClient);
+
+        // putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, createAListOfMessages(getJmsContext(), 3, "msg "));
+
+        final long start = System.nanoTime();
+        final List<SourceRecord> records = connectTask.poll();
+        final long durationMs = (System.nanoTime() - start) / 1_000_000;
+
+        assertThat(durationMs >= 400 && durationMs <= 450).isTrue();
+        assertEquals(0, records.size());
+    }
+
+    @Test
+    public void testPollWithShortMaxPollTime() throws Exception {
+        connectTask = getSourceTaskWithEmptyKafkaOffset();
+        final Map<String, String> config = createExactlyOnceConnectorProperties();
+        config.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
+        config.put("mq.receive.max.poll.time.ms", "50");
+        config.put("mq.message.receive.timeout", "1");
+        config.put("mq.receive.subsequent.timeout.ms", "0");
+        config.put("mq.batch.size", "5000");
+
+        final JMSWorker shared = new JMSWorker();
+        shared.configure(getPropertiesConfig(config));
+        final JMSWorker dedicated = new JMSWorker();
+        dedicated.configure(getPropertiesConfig(config));
+        final SequenceStateClient sequenceStateClient = new SequenceStateClient(DEFAULT_STATE_QUEUE, shared, dedicated);
+        connectTask.start(config, shared, dedicated, sequenceStateClient);
+
+        putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, createAListOfMessages(getJmsContext(), 100, "msg "));
+
+        final List<SourceRecord> records = connectTask.poll();
+
+        assertThat(records.size() < 100);
     }
 }

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

@@ -188,8 +188,11 @@ public class MQSourceConnector extends SourceConnector {
     public static final long CONFIG_RECONNECT_DELAY_MAX_MINIMUM = 10L;
 
     public static final String CONFIG_MAX_POLL_TIME = "mq.receive.max.poll.time.ms";
-    public static final String CONFIG_DOCUMENTATION_MAX_POLL_TIME = "Maximum time (in milliseconds) to spend polling messages before returning a batch to Kafka. "
-            + "If not set or set to 0, polling continues until batch size or receive timeout conditions are met.";
+    public static final String CONFIG_DOCUMENTATION_MAX_POLL_TIME = "Maximum time (in milliseconds) to poll for messages during a single Kafka Connect poll cycle. "
+            + "Acts as a hard upper bound on how long the task will try to accumulate a batch. "
+            + "If set to 0 or not defined, polling continues until either a message receive returns null or the batch size is met. "
+            + "Note: It is recommended to keep this value less than or equal to both 'mq.message.receive.timeout' "
+            + "and 'mq.receive.subsequent.timeout.ms' to avoid unexpected delays due to long blocking receive calls.";
     public static final String CONFIG_DISPLAY_MAX_POLL_TIME = "Max poll time (ms)";
     public static final long CONFIG_MAX_POLL_TIME_DEFAULT = 0L;
 
@@ -615,7 +618,7 @@ null, new ReadableFile(),
                 CONFIG_DISPLAY_MAX_RECEIVE_TIMEOUT);
         CONFIGDEF.define(CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT,
                 ConfigDef.Type.LONG,
-                CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT_DEFAULT,
+                CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT_DEFAULT, ConfigDef.Range.atLeast(CONFIG_SUBSEQUENT_RECEIVE_TIMEOUT_DEFAULT),
                 ConfigDef.Importance.LOW,
                 CONFIG_DOCUMENTATION_SUBSEQUENT_RECEIVE_TIMEOUT,
                 CONFIG_GROUP_MQ,
@@ -656,7 +659,7 @@ null, new ReadableFile(),
                 DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY);
         CONFIGDEF.define(CONFIG_MAX_POLL_TIME,
                 ConfigDef.Type.LONG,
-                CONFIG_MAX_POLL_TIME_DEFAULT,
+                CONFIG_MAX_POLL_TIME_DEFAULT, ConfigDef.Range.atLeast(CONFIG_MAX_POLL_TIME_DEFAULT),
                 ConfigDef.Importance.LOW,
                 CONFIG_DOCUMENTATION_MAX_POLL_TIME,
                 CONFIG_GROUP_MQ,