feat: support of errors.tolerance

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

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

@@ -363,6 +363,8 @@ private List<SourceRecord> internalPoll() throws InterruptedException, JMSRuntim
         final List<SourceRecord> sourceRecordList = messageList.stream()
                 .peek(saveMessageID(msgIds))
                 .map(message -> reader.toSourceRecord(message, sourceQueueConfig.isMqMessageBodyJms(), sourceOffset, sourceQueuePartition))
+                // TODO: comment this for errors.tolerance
+                .filter(sourceRecord -> sourceRecord != null)
                 .collect(Collectors.toList());
 
         // In RE-DELIVER we already have a state on the queue

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;
     }
 }