feat: move out the error handling part to a seperate class

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

Author: Joel-hanson

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

@@ -836,6 +836,7 @@ public void shouldPreserveDlqHeadersWithErrorInformation() throws Exception {
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_TOPIC, "mytopic");
         connectorConfigProps.put(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, "all");
         connectorConfigProps.put(MQSourceConnector.DLQ_TOPIC_NAME_CONFIG, "__dlq.mq.source");
+        connectorConfigProps.put(MQSourceConnector.DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER,
                 "com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
@@ -921,6 +922,7 @@ public void shouldPreserveJmsPropertiesInDlqMessages() throws Exception {
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_TOPIC, "mytopic");
         connectorConfigProps.put(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, "all");
         connectorConfigProps.put(MQSourceConnector.DLQ_TOPIC_NAME_CONFIG, "__dlq.mq.source");
+        connectorConfigProps.put(MQSourceConnector.DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_JMS_PROPERTY_COPY_TO_KAFKA_HEADER, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER,
@@ -1031,6 +1033,7 @@ public void verifyHeadersWithErrorTolerance() throws Exception {
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_TOPIC, "mytopic");
         connectorConfigProps.put(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, "all");
         connectorConfigProps.put(MQSourceConnector.DLQ_TOPIC_NAME_CONFIG, "__dlq.mq.source");
+        connectorConfigProps.put(MQSourceConnector.DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS, "true");
         connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER,
                 "com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
@@ -1109,6 +1112,56 @@ public void verifyHeadersWithErrorTolerance() throws Exception {
                 .toString().contains("com.ibm.eventstreams.connect.mqsource.JMSWorker.toSourceRecord")).isTrue();
     }
 
+    @Test
+    public void verifyHeadersWithErrorTolerance_WithDLQHeaderContextDisabled() throws Exception {
+        connectTask = getSourceTaskWithEmptyKafkaOffset();
+
+        final Map<String, String> connectorConfigProps = createDefaultConnectorProperties();
+        connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_TOPIC, "mytopic");
+        connectorConfigProps.put(ConnectorConfig.ERRORS_TOLERANCE_CONFIG, "all");
+        connectorConfigProps.put(MQSourceConnector.DLQ_TOPIC_NAME_CONFIG, "__dlq.mq.source");
+        connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS, "true");
+        connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER,
+                "com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
+        connectorConfigProps.put(MQSourceConnector.CONFIG_NAME_MQ_JMS_PROPERTY_COPY_TO_KAFKA_HEADER, "true");
+
+        connectTask.start(connectorConfigProps);
+
+        final TextMessage message = getJmsContext().createTextMessage("Invalid JSON message");
+        message.setStringProperty("teststring", "myvalue");
+        message.setIntProperty("volume", 11);
+        message.setDoubleProperty("decimalmeaning", 42.0);
+
+        // Both invalid and valid messages are received
+        final List<Message> testMessages = Arrays.asList(
+                message, // Poison message
+                getJmsContext().createTextMessage("{ \"i\": 0 }") // Valid message
+        );
+        putAllMessagesToQueue(DEFAULT_SOURCE_QUEUE, testMessages);
+
+        final List<SourceRecord> processedRecords = connectTask.poll();
+
+        assertThat(processedRecords).hasSize(2);
+
+        final SourceRecord dlqRecord = processedRecords.get(0);
+        assertThat(dlqRecord.topic()).isEqualTo("__dlq.mq.source");
+
+        final Headers headers = dlqRecord.headers();
+
+        // Actual headers
+        assertThat(headers.lastWithName("teststring").value()).isEqualTo("myvalue");
+        assertThat(headers.lastWithName("volume").value()).isEqualTo("11");
+        assertThat(headers.lastWithName("decimalmeaning").value()).isEqualTo("42.0");
+
+        assertThat(headers.lastWithName("__connect.errors.topic")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.class.name")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.exception.message")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.timestamp")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.cause.message")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.cause.class")).isNull();
+        assertThat(headers.lastWithName("__connect.errors.exception.stacktrace")).isNull();
+    }
+
     @Test
     public void verifyLoggingWarningWithErrorTolerance() throws Exception {
         connectTask = getSourceTaskWithEmptyKafkaOffset();

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

@@ -196,11 +196,6 @@ public class MQSourceConnector extends SourceConnector {
     public static final String DLQ_TOPIC_DEFAULT = "";
     private static final String DLQ_TOPIC_DISPLAY = "Dead Letter Queue Topic Name";
 
-    public static final String DLQ_TOPIC_REPLICATION_FACTOR_CONFIG = DLQ_PREFIX + "topic.replication.factor";
-    private static final String DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DOC = "Replication factor used to create the dead letter queue topic when it doesn't already exist.";
-    public static final short DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DEFAULT = 3;
-    private static final String DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DISPLAY = "Dead Letter Queue Topic Replication Factor";
-
     public static final String DLQ_CONTEXT_HEADERS_ENABLE_CONFIG = DLQ_PREFIX + "context.headers.enable";
     public static final boolean DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT = false;
     public static final String DLQ_CONTEXT_HEADERS_ENABLE_DOC = "If true, add headers containing error context to the messages " +
@@ -645,20 +640,12 @@ null, new ReadableFile(),
                 CONFIG_GROUP_MQ, 29,
                 Width.MEDIUM,
                 DLQ_TOPIC_DISPLAY);
-        CONFIGDEF.define(DLQ_TOPIC_REPLICATION_FACTOR_CONFIG,
-                Type.SHORT,
-                DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DEFAULT,
-                Importance.MEDIUM,
-                DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DOC,
-                CONFIG_GROUP_MQ, 30,
-                Width.MEDIUM,
-                DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DISPLAY);
         CONFIGDEF.define(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG,
                 Type.BOOLEAN,
                 DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT,
                 Importance.MEDIUM,
                 DLQ_CONTEXT_HEADERS_ENABLE_DOC,
-                CONFIG_GROUP_MQ, 31,
+                CONFIG_GROUP_MQ, 30,
                 Width.MEDIUM,
                 DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY);