test: additional tests

Signed-off-by: Dale Lane <dale.lane@uk.ibm.com>

Author: dalelane

…s/connect/mqsink/SimpleMQSinkTaskIT.java …streams/connect/mqsink/MQSinkTaskIT.javasrc/integration/java/com/ibm/eventstreams/connect/mqsink/SimpleMQSinkTaskIT.java renamed to src/integration/java/com/ibm/eventstreams/connect/mqsink/MQSinkTaskIT.java src/integration/java/com/ibm/eventstreams/connect/mqsink/SimpleMQSinkTaskIT.java renamed to src/integration/java/com/ibm/eventstreams/connect/mqsink/MQSinkTaskIT.java

@@ -1,6 +1,7 @@
 package com.ibm.eventstreams.connect.mqsink;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -16,10 +17,11 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.Test;
 
-public class SimpleMQSinkTaskIT extends AbstractJMSContextIT {
+public class MQSinkTaskIT extends AbstractJMSContextIT {
 
     private static final String TOPIC = "SINK.TOPIC.NAME";
     private static final int PARTITION = 3;
@@ -46,6 +48,35 @@ private Map<String, String> createDefaultConnectorProperties() {
     }
 
 
+    @Test
+    public void verifyUnsupportedReplyQueueName() {
+        Map<String, String> connectorConfigProps = createDefaultConnectorProperties();
+        connectorConfigProps.put("mq.message.builder", "com.ibm.eventstreams.connect.mqsink.builders.DefaultMessageBuilder");
+        connectorConfigProps.put("mq.reply.queue", "queue://QM2/Q2?persistence=2&priority=5");
+
+        MQSinkTask newConnectTask = new MQSinkTask();
+        ConnectException exc = assertThrows(ConnectException.class, () -> {
+            newConnectTask.start(connectorConfigProps);
+        });
+
+        assertEquals("Reply-to queue URI must not contain properties", exc.getMessage());
+    }
+
+    @Test
+    public void verifyUnsupportedKeyHeader() {
+        Map<String, String> connectorConfigProps = createDefaultConnectorProperties();
+        connectorConfigProps.put("mq.message.builder", "com.ibm.eventstreams.connect.mqsink.builders.DefaultMessageBuilder");
+        connectorConfigProps.put("mq.message.builder.key.header", "hello");
+
+        MQSinkTask newConnectTask = new MQSinkTask();
+        ConnectException exc = assertThrows(ConnectException.class, () -> {
+            newConnectTask.start(connectorConfigProps);
+        });
+
+        assertEquals("Unsupported MQ message builder key header value", exc.getMessage());
+    }
+
+
     @Test
     public void verifyStringMessages() throws JMSException {
         MQSinkTask newConnectTask = new MQSinkTask();
@@ -207,15 +238,15 @@ public void verifyJsonWithDefaultBuilder() throws JMSException {
         connectorConfigProps.put("mq.message.builder.value.converter", "org.apache.kafka.connect.json.JsonConverter");
         connectorConfigProps.put("mq.message.builder", "com.ibm.eventstreams.connect.mqsink.builders.DefaultMessageBuilder");
 
-        verifyMessageConversion(connectorConfigProps, Schema.STRING_SCHEMA, "\"ABC\"", "\"ABC\"");
+        verifyMessageConversion(connectorConfigProps, Schema.STRING_SCHEMA, "ABC", "ABC");
     }
     @Test
     public void verifyJsonWithJsonBuilder() throws JMSException {
         Map<String, String> connectorConfigProps = createDefaultConnectorProperties();
         connectorConfigProps.put("mq.message.builder.value.converter", "org.apache.kafka.connect.json.JsonConverter");
         connectorConfigProps.put("mq.message.builder", "com.ibm.eventstreams.connect.mqsink.builders.JsonMessageBuilder");
 
-        verifyMessageConversion(connectorConfigProps, Schema.STRING_SCHEMA, "\"ABC\"", "\"\\\"ABC\\\"\"");
+        verifyMessageConversion(connectorConfigProps, Schema.STRING_SCHEMA, "ABC", "\"ABC\"");
     }