feat: Update readme with the new config values

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

Author: Joel-hanson

.github/ISSUE_TEMPLATE/BUG-REPORT.yml

@@ -57,7 +57,7 @@ body:
       label: Version
       description: What version of our software are you running?
       options:
-        - 2.3.0 (Default)
+        - 2.4.0 (Default)
         - 1.3.5
         - older (<1.3.5)
     validations:

README.md

@@ -20,7 +20,7 @@
     <groupId>com.ibm.eventstreams.connect</groupId>
     <artifactId>kafka-connect-mq-source</artifactId>
     <packaging>jar</packaging>
-    <version>2.3.0</version>
+    <version>2.4.0</version>
     <name>kafka-connect-mq-source</name>
     <organization>
         <name>IBM Corporation</name>

pom.xml

@@ -857,7 +857,7 @@ public void shouldPreserveDlqHeadersWithErrorInformation() throws Exception {
         final Headers headers = dlqRecord.headers();
         assertThat(headers.lastWithName("__connect.errors.topic").value())
                 .isEqualTo("mytopic");
-        assertThat(headers.lastWithName("__connect.errors.class.name").value())
+        assertThat(headers.lastWithName("__connect.errors.exception.class.name").value())
                 .isEqualTo("org.apache.kafka.connect.errors.DataException");
         assertThat(headers.lastWithName("__connect.errors.exception.message").value())
                 .isEqualTo("Converting byte[] to Kafka Connect data failed due to serialization error: ");
@@ -1094,7 +1094,7 @@ public void verifyHeadersWithErrorTolerance() throws Exception {
         // [truncated], schema=Schema{STRING})])
         assertThat(headers.lastWithName("__connect.errors.topic").value())
                 .isEqualTo("mytopic");
-        assertThat(headers.lastWithName("__connect.errors.class.name").value())
+        assertThat(headers.lastWithName("__connect.errors.exception.class.name").value())
                 .isEqualTo("org.apache.kafka.connect.errors.DataException");
         assertThat(headers.lastWithName("__connect.errors.exception.message").value())
                 .isEqualTo("Converting byte[] to Kafka Connect data failed due to serialization error: ");

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

@@ -209,7 +209,7 @@ public class MQSourceConnector extends SourceConnector {
         CONFIG_VALUE_MQ_CLIENT_RECONNECT_OPTION_DISABLED.toLowerCase(Locale.ENGLISH)
     };
 
-    public static String version = "2.3.0";
+    public static String version = "2.4.0";
 
     private Map<String, String> configProps;
 

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

@@ -17,7 +17,6 @@
 import org.apache.kafka.connect.header.ConnectHeaders;
 import org.apache.kafka.connect.header.Headers;
 import org.apache.kafka.connect.runtime.ConnectorConfig;
-import org.apache.kafka.connect.runtime.errors.DeadLetterQueueReporter;
 import org.apache.kafka.connect.runtime.errors.ToleranceType;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.slf4j.Logger;
@@ -35,12 +34,6 @@ public class ErrorHandler {
 
     public static final String HEADER_PREFIX = "__connect.errors.";
     public static final String ERROR_HEADER_ORIG_TOPIC = HEADER_PREFIX + "topic";
-    public static final String ERROR_HEADER_ORIG_PARTITION = HEADER_PREFIX + "partition";
-    public static final String ERROR_HEADER_ORIG_OFFSET = HEADER_PREFIX + "offset";
-    public static final String ERROR_HEADER_CONNECTOR_NAME = HEADER_PREFIX + "connector.name";
-    public static final String ERROR_HEADER_TASK_ID = HEADER_PREFIX + "task.id";
-    public static final String ERROR_HEADER_STAGE = HEADER_PREFIX + "stage";
-    public static final String ERROR_HEADER_EXECUTING_CLASS = HEADER_PREFIX + "class.name";
     public static final String ERROR_HEADER_EXCEPTION = HEADER_PREFIX + "exception.class.name";
     public static final String ERROR_HEADER_EXCEPTION_MESSAGE = HEADER_PREFIX + "exception.message";
     public static final String ERROR_HEADER_EXCEPTION_STACK_TRACE = HEADER_PREFIX + "exception.stacktrace";
@@ -113,7 +106,6 @@ private void initializeErrorTolerance(final Map<String, String> props) {
             dlqTopic = props.get(MQSourceConnector.DLQ_TOPIC_NAME_CONFIG);
             if (dlqTopic != null && !dlqTopic.isEmpty()) {
                 dlqTopic = dlqTopic.trim();
-                // TODO: Check if DLQ topic exists
             }
 
             queueName = props.get(MQSourceConnector.CONFIG_NAME_MQ_QUEUE);
@@ -282,9 +274,9 @@ private Headers createErrorHeaders(final Message message, final String originalT
         }
 
         // Basic error information
-        headers.addString(DeadLetterQueueReporter.ERROR_HEADER_ORIG_TOPIC, originalTopic);
-        headers.addString(DeadLetterQueueReporter.ERROR_HEADER_EXECUTING_CLASS, exception.getClass().getName());
-        headers.addString(DeadLetterQueueReporter.ERROR_HEADER_EXCEPTION_MESSAGE, exception.getMessage());
+        headers.addString(ERROR_HEADER_ORIG_TOPIC, originalTopic);
+        headers.addString(ERROR_HEADER_EXCEPTION, exception.getClass().getName());
+        headers.addString(ERROR_HEADER_EXCEPTION_MESSAGE, exception.getMessage());
 
         try {
             headers.addString(ERROR_HEADER_JMS_MESSAGE_ID, message.getJMSMessageID());
@@ -298,14 +290,14 @@ private Headers createErrorHeaders(final Message message, final String originalT
 
         // Add cause if available
         if (exception.getCause() != null) {
-            headers.addString(ERROR_HEADER_EXCEPTION_CAUSE_MESSAGE, exception.getCause().getMessage());
             headers.addString(ERROR_HEADER_EXCEPTION_CAUSE_CLASS, exception.getCause().getClass().getName());
+            headers.addString(ERROR_HEADER_EXCEPTION_CAUSE_MESSAGE, exception.getCause().getMessage());
         }
 
         // Add first few lines of stack trace (full stack trace might be too large)
         final String stackTrace = getStackTrace(exception);
         if (stackTrace != null) {
-            headers.addString(DeadLetterQueueReporter.ERROR_HEADER_EXCEPTION_STACK_TRACE, stackTrace);
+            headers.addString(ERROR_HEADER_EXCEPTION_STACK_TRACE, stackTrace);
         }
 
         return headers;