fix: rollback messages to MQ if they cannot be transformed

Some JMS message types cannot be processed by the record builder.
To avoid losing these messages, they should be rolled back to the
MQ queue that they came from.

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

Author: dalelane

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

@@ -15,6 +15,7 @@
  */
 package com.ibm.eventstreams.connect.mqsource;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
 
@@ -23,6 +24,7 @@
 import javax.jms.JMSContext;
 import javax.jms.JMSException;
 import javax.jms.Message;
+import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 
@@ -143,4 +145,48 @@ public void putAllMessagesToQueue(String queueName, List<Message> messages) thro
 
         connection.close();
     }
+
+
+    /**
+     * Gets all messages from the specified MQ queue. Used in tests to
+     *  verify what is left on the test queue
+     */
+    public List<Message> getAllMessagesFromQueue(String queueName) throws JMSException {
+        Connection connection = null;
+        Session session = null;
+        Destination destination = null;
+        MessageConsumer consumer = null;
+
+        JmsFactoryFactory ff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);
+
+        JmsConnectionFactory cf = ff.createConnectionFactory();
+        cf.setStringProperty(WMQConstants.WMQ_HOST_NAME, "localhost");
+        cf.setIntProperty(WMQConstants.WMQ_PORT, getMQPort());
+        cf.setStringProperty(WMQConstants.WMQ_CHANNEL, getChannelName());
+        cf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
+        cf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, getQmgrName());
+        cf.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, false);
+
+        connection = cf.createConnection();
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        destination = session.createQueue(queueName);
+        consumer = session.createConsumer(destination);
+
+        connection.start();
+
+        List<Message> messages = new ArrayList<>();
+        Message message;
+        do {
+            message = consumer.receiveNoWait();
+            if (message != null) {
+                messages.add(message);
+            }
+        }
+        while (message != null);
+
+        connection.close();
+
+        return messages;
+    }
 }

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

@@ -19,17 +19,22 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.jms.MapMessage;
 import javax.jms.Message;
+import javax.jms.MessageFormatException;
 import javax.jms.TextMessage;
 
 import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.junit.Test;
 
@@ -246,6 +251,57 @@ public void verifyMessageBatchGroupCommits() throws Exception {
 
 
 
+    @Test
+    public void verifyMessageBatchRollback() throws Exception {
+        MQSourceTask newConnectTask = new MQSourceTask();
+
+        Map<String, String> connectorConfigProps = createDefaultConnectorProperties();
+        connectorConfigProps.put("mq.message.body.jms", "true");
+        connectorConfigProps.put("mq.record.builder", "com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder");
+        connectorConfigProps.put("mq.batch.size", "10");
+
+        newConnectTask.start(connectorConfigProps);
+
+        // Test overview:
+        //
+        // messages 01-15 - valid messages
+        // message  16    - a message that the builder can't process
+        // messages 17-30 - valid messages
+
+        List<Message> messages = new ArrayList<>();
+        for (int i = 1; i <= 15; i++) {
+            messages.add(getJmsContext().createTextMessage("message " + i));
+        }
+        MapMessage invalidMessage = getJmsContext().createMapMessage();
+        invalidMessage.setString("test", "builder cannot convert this");
+        messages.add(invalidMessage);
+        for (int i = 17; i <= 30; i++) {
+            messages.add(getJmsContext().createTextMessage("message " + i));
+        }
+        putAllMessagesToQueue(MQ_QUEUE, messages);
+
+        List<SourceRecord> kafkaMessages;
+
+        // first batch should successfully retrieve messages 01-10
+        kafkaMessages = newConnectTask.poll();
+        assertEquals(10, kafkaMessages.size());
+        newConnectTask.commit();
+        newConnectTask.commit();
+
+        // second batch (11-20) should fail because of message 16
+        ConnectException exc = assertThrows(ConnectException.class, () -> {
+            newConnectTask.poll();
+        });
+        assertTrue(exc.getMessage().equals("Unsupported JMS message type"));
+
+        // there should be 20 messages left on the MQ queue (messages 11-30)
+        newConnectTask.stop();
+        List<Message> remainingMQMessages = getAllMessagesFromQueue(MQ_QUEUE);
+        assertEquals(20, remainingMQMessages.size());
+    }
+
+
+
     @Test
     public void verifyMessageIdAsKey() throws Exception {
         MQSourceTask newConnectTask = new MQSourceTask();

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

@@ -292,13 +292,31 @@ public SourceRecord receive(boolean wait) {
         }
         catch (ConnectException exc) {
             log.error("Connect exception {}", exc);
+            attemptRollback();
             throw exc;
         }
 
         log.trace("[{}]  Exit {}.receive, retval={}", Thread.currentThread().getId(), this.getClass().getName(), sr);
         return sr;
     }
 
+
+    /**
+     * Returns messages got from the MQ queue. Called if the builder has failed to transform the
+     *  messages and return them to Connect for producing to Kafka.
+     */
+    private void attemptRollback() {
+        log.trace("[{}] Entry {}.attemptRollback", Thread.currentThread().getId(), this.getClass().getName());
+        try {
+            jmsCtxt.rollback();
+        }
+        catch (JMSRuntimeException jmsExc) {
+            log.error("rollback failed {}", jmsExc);
+        }
+        log.trace("[{}]  Exit {}.attemptRollback", Thread.currentThread().getId(), this.getClass().getName());
+    }
+
+
     /**
      * Commits the current transaction. If the current transaction contains a message that could not
      * be processed, the transaction is "in peril" and is rolled back instead to avoid data loss.