feat: support of errors.tolerance

- Added a IT for error tolerance
- Added conditions with in JSONRecordBuilder for error tolerance

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

Author: Joel-hanson

pom.xml

@@ -54,6 +54,12 @@
             <version>3.7.1</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-runtime</artifactId>
+            <version>3.7.1</version>
+            <scope>provided</scope>
+        </dependency>
 
         <dependency>
             <groupId>javax.jms</groupId>

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

@@ -20,13 +20,15 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import javax.jms.BytesMessage;
 import javax.jms.MapMessage;
 import javax.jms.TextMessage;
 
+import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.junit.Test;
 
@@ -102,4 +104,32 @@ public void buildFromJmsMapMessage() throws Exception {
         // verify the exception
         assertEquals("Unsupported JMS message type", exc.getMessage());
     }
+
+    @Test
+    public void buildFromJmsTestJsonError() throws Exception {
+        // create MQ message
+        final TextMessage message = getJmsContext().createTextMessage("Not a valid json string");
+
+        // 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());
+    }
+
+    @Test
+    public void buildFromJmsTestErrorTolerance() throws Exception {
+        // create MQ message
+        final TextMessage message = getJmsContext().createTextMessage("Not a valid json string");
+
+        // use the builder to convert it to a Kafka record
+        final JsonRecordBuilder builder = new JsonRecordBuilder();
+        final HashMap<String, String> config = new HashMap<String, String>();
+        config.put("errors.tolerance", "all");
+        builder.configure(config);
+        final SourceRecord record = builder.toSourceRecord(getJmsContext(), topic, isJMS, message);
+        assertNull(record.key());
+        assertNull(record.valueSchema());
+        verifyJsonMap((Map<?, ?>) record.value());
+    }
 }

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

@@ -24,6 +24,9 @@
 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;
@@ -37,6 +40,7 @@
 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;
@@ -108,6 +112,7 @@ 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";
@@ -146,6 +151,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);
@@ -362,7 +368,17 @@ private List<SourceRecord> internalPoll() throws InterruptedException, JMSRuntim
         final ArrayList<String> msgIds = new ArrayList<>();
         final List<SourceRecord> sourceRecordList = messageList.stream()
                 .peek(saveMessageID(msgIds))
-                .map(message -> reader.toSourceRecord(message, sourceQueueConfig.isMqMessageBodyJms(), sourceOffset, sourceQueuePartition))
+                .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)
                 .collect(Collectors.toList());
 
         // In RE-DELIVER we already have a state on the queue
@@ -585,4 +601,9 @@ 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);
+    }
 }

src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java

@@ -27,6 +27,7 @@
 import javax.jms.JMSContext;
 import javax.jms.JMSException;
 import javax.jms.Message;
+
 import java.util.Map;
 import java.util.Optional;
 

src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java

@@ -18,6 +18,9 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
 import javax.jms.BytesMessage;
 import javax.jms.JMSContext;
 import javax.jms.JMSException;
@@ -26,7 +29,8 @@
 
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.json.JsonConverter;
-
+import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.kafka.connect.runtime.errors.ToleranceType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,6 +42,7 @@ public class JsonRecordBuilder extends BaseRecordBuilder {
     private static final Logger log = LoggerFactory.getLogger(JsonRecordBuilder.class);
 
     private JsonConverter converter;
+    private ToleranceType toleranceType;
 
     public JsonRecordBuilder() {
         log.info("Building records using com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
@@ -51,6 +56,13 @@ public JsonRecordBuilder() {
         converter.configure(m, false);
     }
 
+    @Override
+    public void configure(final Map<String, String> props) {
+        super.configure(props);
+        final String errorTolerance = props.getOrDefault(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, ConnectorConfig.ERRORS_TOLERANCE_DEFAULT.toString());
+        toleranceType = ToleranceType.valueOf(errorTolerance.toUpperCase(Locale.ROOT));
+    }
+
     /**
      * Gets the value schema to use for the Kafka Connect SourceRecord.
      *
@@ -78,6 +90,12 @@ public SchemaAndValue getValue(final JMSContext context, final String topic, fin
             throw new RecordBuilderException("Unsupported JMS message type");
         }
 
-        return converter.toConnectData(topic, payload);
+        try {
+            return converter.toConnectData(topic, payload);
+        } catch (final Exception e) {
+            if (toleranceType == ToleranceType.NONE)
+                throw e;
+        }
+        return null;
     }
 }