Merge pull request #99 from dalelane/jms-rollback

fix: rollback messages to MQ if they cannot be transformed

Author: gmcrobert

Dockerfile

@@ -10,7 +10,7 @@ COPY --chown=esuser:esgroup --from=builder /opt/kafka/libs/ /opt/kafka/libs/
 COPY --chown=esuser:esgroup --from=builder /opt/kafka/config/connect-distributed.properties /opt/kafka/config/
 COPY --chown=esuser:esgroup --from=builder /opt/kafka/config/connect-log4j.properties /opt/kafka/config/
 RUN mkdir /opt/kafka/logs && chown esuser:esgroup /opt/kafka/logs
-COPY --chown=esuser:esgroup target/kafka-connect-mq-source-1.3.1-jar-with-dependencies.jar /opt/kafka/libs/
+COPY --chown=esuser:esgroup target/kafka-connect-mq-source-1.3.2-jar-with-dependencies.jar /opt/kafka/libs/
 
 WORKDIR /opt/kafka
 

pom.xml

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

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.After;
 import org.junit.Test;
@@ -259,6 +264,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 {
         connectTask = 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.

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

@@ -135,7 +135,7 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_DOCUMENTATION_TOPIC = "The name of the target Kafka topic.";
     public static final String CONFIG_DISPLAY_TOPIC = "Target Kafka topic";
 
-    public static String VERSION = "1.3.1";
+    public static String VERSION = "1.3.2";
 
     private Map<String, String> configProps;