feat: handle poison messages from base builder class

- Handle poison messages from base builder class
- Poison message is handle when `errors.tolerance` is set to all
- Option to configure deadletter queue for poison messages
- New UTs for testing errors tolerance behaviour

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

Author: Joel-hanson

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

@@ -94,7 +94,7 @@ protected JMSContext getJmsContext() throws Exception {
         return jmsContext;
     }
 
-    protected Map<String, String> getDefaultConnectorProperties() {
+    public static Map<String, String> getDefaultConnectorProperties() {
         final Map<String, String> props = new HashMap<>();
         props.put("mq.queue.manager", QMGR_NAME);
         props.put("mq.connection.mode", "client");
@@ -106,7 +106,7 @@ protected Map<String, String> getDefaultConnectorProperties() {
         return props;
     }
 
-    public AbstractConfig getPropertiesConfig(Map<String, String> props) {
+    public AbstractConfig getPropertiesConfig(final Map<String, String> props) {
         final AbstractConfig connectorConfig = new AbstractConfig(MQSourceConnector.CONFIGDEF, props);
         return connectorConfig;
     }

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

@@ -20,7 +20,6 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -113,8 +112,7 @@ public void buildFromJmsTestJsonError() throws Exception {
         // use the builder to convert it to a Kafka record
         final JsonRecordBuilder builder = new JsonRecordBuilder();
         final DataException exec = assertThrows(DataException.class, () -> builder.toSourceRecord(getJmsContext(), topic, isJMS, message));
-        System.out.println(exec);
-        System.out.println(exec.getMessage());
+        assertEquals("Converting byte[] to Kafka Connect data failed due to serialization error: ", exec.getMessage());
     }
 
     @Test
@@ -124,12 +122,13 @@ public void buildFromJmsTestErrorTolerance() throws Exception {
 
         // use the builder to convert it to a Kafka record
         final JsonRecordBuilder builder = new JsonRecordBuilder();
-        final HashMap<String, String> config = new HashMap<String, String>();
+        final Map<String, String> config = AbstractJMSContextIT.getDefaultConnectorProperties();
         config.put("errors.tolerance", "all");
+        config.put("mq.message.body.jms", "true");
+        config.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
+
         builder.configure(config);
         final SourceRecord record = builder.toSourceRecord(getJmsContext(), topic, isJMS, message);
-        assertNull(record.key());
-        assertNull(record.valueSchema());
-        verifyJsonMap((Map<?, ?>) record.value());
+        assertNull(record);
     }
 }

src/integration/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilderIT.java

@@ -181,6 +181,14 @@ public class MQSourceConnector extends SourceConnector {
     public static final long CONFIG_RECONNECT_DELAY_MAX_DEFAULT = 8192L;
     public static final long CONFIG_RECONNECT_DELAY_MAX_MINIMUM = 10L;
 
+    public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name";
+    public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_DISPLAY = "Output errors to the dead letter queue";
+    public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_DEFAULT = "";
+    public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_DOC = "Whether to output conversion errors to the dead letter queue. "
+            + "Stops poison messages when using schemas, any message will be outputted as extended json on the specified topic. "
+            + "By default messages are not outputted to the dead letter queue. "
+            + "Also requires `errors.tolerance=all`.";
+
     // Define valid reconnect options
     public static final String[] CONFIG_VALUE_MQ_VALID_RECONNECT_OPTIONS = {
         CONFIG_VALUE_MQ_CLIENT_RECONNECT_OPTION_ASDEF,

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

@@ -24,9 +24,6 @@
 import com.ibm.eventstreams.connect.mqsource.util.QueueConfig;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.connect.errors.ConnectException;
-import org.apache.kafka.connect.errors.DataException;
-import org.apache.kafka.connect.runtime.ConnectorConfig;
-import org.apache.kafka.connect.runtime.errors.ToleranceType;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
 import org.slf4j.Logger;
@@ -40,7 +37,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
@@ -112,7 +108,6 @@ public class MQSourceTask extends SourceTask {
 
     private int startActionPollLimit = 300; // This is a 5 minute time out on the initial start procedure
     private AtomicInteger startActionPollCount = new AtomicInteger(0);
-    private boolean tolerateErrors = false;
 
     private final static String OFFSET_IDENTIFIER = "sequence-id";
     private final static String SOURCE_PARTITION_IDENTIFIER = "source";
@@ -151,7 +146,7 @@ public void start(final Map<String, String> props) {
         final JMSWorker reader = new JMSWorker();
         JMSWorker dedicated = null;
         SequenceStateClient client = null;
-        tolerateErrors = tolerateErrors(props);
+
         if (MQSourceConnector.configSupportsExactlyOnce(props)) {
             dedicated = new JMSWorker();
             client = new SequenceStateClient(props.get(CONFIG_NAME_MQ_EXACTLY_ONCE_STATE_QUEUE), reader, dedicated);
@@ -368,17 +363,7 @@ private List<SourceRecord> internalPoll() throws InterruptedException, JMSRuntim
         final ArrayList<String> msgIds = new ArrayList<>();
         final List<SourceRecord> sourceRecordList = messageList.stream()
                 .peek(saveMessageID(msgIds))
-                .map(message -> {
-                    try {
-                        return reader.toSourceRecord(message, sourceQueueConfig.isMqMessageBodyJms(), sourceOffset, sourceQueuePartition);
-                    } catch (final DataException dataException) {
-                        if (!tolerateErrors) {
-                            throw dataException;
-                        }
-                    }
-                    return null;
-                })
-                .filter(sourceRecord -> sourceRecord != null)
+                .map(message -> reader.toSourceRecord(message, sourceQueueConfig.isMqMessageBodyJms(), sourceOffset, sourceQueuePartition))
                 .collect(Collectors.toList());
 
         // In RE-DELIVER we already have a state on the queue
@@ -601,9 +586,4 @@ private void closeAllWorkers() {
             dedicated.close();
         }
     }
-
-    public static boolean tolerateErrors(final Map<String, String> sfConnectorConfig) {
-        final String errorsTolerance = sfConnectorConfig.getOrDefault(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, ToleranceType.NONE.toString()).toUpperCase(Locale.ROOT);
-        return ToleranceType.valueOf(errorsTolerance).equals(ToleranceType.ALL);
-    }
 }