How to simulate message re-delivery in a JMS AUTO_ACKNOWLEDGE session script?

Question

In the following test, I am trying to simulate the following scenario:

• The message queue starts.
• The user created to fail when processing messages is launched.
• A message is displayed.
• The user starts processing the message.
• During processing, an exception is thrown to simulate a message processing failure. The failed consumer is stopped.
• Another user starts up with the intention of receiving a redirected message.

But my test fails, and the message is not forwarded to the new consumer. I would appreciate any hints of this.

MessageProcessingFailureAndReprocessingTest.java

@ContextConfiguration(locations="com.prototypo.queue.MessageProcessingFailureAndReprocessingTest$ContextConfig", loader=JavaConfigContextLoader.class) public class MessageProcessingFailureAndReprocessingTest extends AbstractJUnit4SpringContextTests { @Autowired private FailureReprocessTestScenario testScenario; @Before public void setUp() { testScenario.start(); } @After public void tearDown() throws Exception { testScenario.stop(); } @Test public void should_reprocess_task_after_processing_failure() { try { Thread.sleep(20*1000); assertThat(testScenario.succeedingWorker.processedTasks, is(Arrays.asList(new String[]{ "task-1", }))); } catch (InterruptedException e) { fail(); } } @Configurable public static class FailureReprocessTestScenario { @Autowired public BrokerService broker; @Autowired public MockTaskProducer mockTaskProducer; @Autowired public FailingWorker failingWorker; @Autowired public SucceedingWorker succeedingWorker; @Autowired public TaskScheduler scheduler; public void start() { Date now = new Date(); scheduler.schedule(new Runnable() { public void run() { failingWorker.start(); } }, now); Date after1Seconds = new Date(now.getTime() + 1*1000); scheduler.schedule(new Runnable() { public void run() { mockTaskProducer.produceTask(); } }, after1Seconds); Date after2Seconds = new Date(now.getTime() + 2*1000); scheduler.schedule(new Runnable() { public void run() { failingWorker.stop(); succeedingWorker.start(); } }, after2Seconds); } public void stop() throws Exception { succeedingWorker.stop(); broker.stop(); } } @Configuration @ImportResource(value={"classpath:applicationContext-jms.xml", "classpath:applicationContext-task.xml"}) public static class ContextConfig { @Autowired private ConnectionFactory jmsFactory; @Bean public FailureReprocessTestScenario testScenario() { return new FailureReprocessTestScenario(); } @Bean public MockTaskProducer mockTaskProducer() { return new MockTaskProducer(); } @Bean public FailingWorker failingWorker() { TaskListener listener = new TaskListener(); FailingWorker worker = new FailingWorker(listenerContainer(listener)); listener.setProcessor(worker); return worker; } @Bean public SucceedingWorker succeedingWorker() { TaskListener listener = new TaskListener(); SucceedingWorker worker = new SucceedingWorker(listenerContainer(listener)); listener.setProcessor(worker); return worker; } private DefaultMessageListenerContainer listenerContainer(TaskListener listener) { DefaultMessageListenerContainer listenerContainer = new DefaultMessageListenerContainer(); listenerContainer.setConnectionFactory(jmsFactory); listenerContainer.setDestinationName("tasksQueue"); listenerContainer.setMessageListener(listener); listenerContainer.setAutoStartup(false); listenerContainer.initialize(); return listenerContainer; } } public static class FailingWorker implements TaskProcessor { private Logger LOG = Logger.getLogger(FailingWorker.class.getName()); private final DefaultMessageListenerContainer listenerContainer; public FailingWorker(DefaultMessageListenerContainer listenerContainer) { this.listenerContainer = listenerContainer; } public void start() { LOG.info("FailingWorker.start()"); listenerContainer.start(); } public void stop() { LOG.info("FailingWorker.stop()"); listenerContainer.stop(); } @Override public void processTask(Object task) { LOG.info("FailingWorker.processTask(" + task + ")"); try { Thread.sleep(1*1000); throw Throwables.propagate(new Exception("Simulate task processing failure")); } catch (InterruptedException e) { LOG.log(Level.SEVERE, "Unexpected interruption exception"); } } } public static class SucceedingWorker implements TaskProcessor { private Logger LOG = Logger.getLogger(SucceedingWorker.class.getName()); private final DefaultMessageListenerContainer listenerContainer; public final List<String> processedTasks; public SucceedingWorker(DefaultMessageListenerContainer listenerContainer) { this.listenerContainer = listenerContainer; this.processedTasks = new ArrayList<String>(); } public void start() { LOG.info("SucceedingWorker.start()"); listenerContainer.start(); } public void stop() { LOG.info("SucceedingWorker.stop()"); listenerContainer.stop(); } @Override public void processTask(Object task) { LOG.info("SucceedingWorker.processTask(" + task + ")"); try { TextMessage taskText = (TextMessage) task; processedTasks.add(taskText.getText()); } catch (JMSException e) { LOG.log(Level.SEVERE, "Unexpected exception during task processing"); } } } } 

TaskListener.java

 public class TaskListener implements MessageListener { private TaskProcessor processor; @Override public void onMessage(Message message) { processor.processTask(message); } public void setProcessor(TaskProcessor processor) { this.processor = processor; } } 

MockTaskProducer.java

 @Configurable public class MockTaskProducer implements ApplicationContextAware { private Logger LOG = Logger.getLogger(MockTaskProducer.class.getName()); @Autowired private JmsTemplate jmsTemplate; private Destination destination; private int taskCounter = 0; public void produceTask() { LOG.info("MockTaskProducer.produceTask(" + taskCounter + ")"); taskCounter++; jmsTemplate.send(destination, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { TextMessage message = session.createTextMessage("task-" + taskCounter); return message; } }); } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { destination = applicationContext.getBean("tasksQueue", Destination.class); } } 

Answer 1

Apparently, the source of the documentation that I looked at yesterday Creating reliable JMS applications is misleading (or maybe I got it wrong). Especially that excerpt:

Until the JMS message is acknowledged, it is not deemed to be successfully consumed. Successful message consumption usually occurs in three stages.

• The client receives a message.
• The client is processing the message.
• The message is confirmed. Acknowledgment is initiated either by the JMS provider or by the client, depending on the session confirmation mode.

I assumed that AUTO_ACKNOWLEDGE does just that - confirmed the message after the listener method returns the result. But according to the JMS specification, this is a little different, and Spring listener containers, as expected, are not trying to change the behavior from the JMS specification. This is what javadoc AbstractMessageListenerContainer has to say - I emphasized important suggestions:

The listener container offers the following acknowledgment of the Parameters message:

• "sessionAcknowledgeMode" is set to "AUTO_ACKNOWLEDGE" (default): Automatic confirmation of the message before the listener executes; no redelivery in case of exception.
• "sessionAcknowledgeMode" is set to "CLIENT_ACKNOWLEDGE": automatic confirmation of a message after a successful listener; no redelivery in case of exception.
• "sessionAcknowledgeMode" is set to "DUPS_OK_ACKNOWLEDGE": lazy confirmation of the message during or after the listener is executed; potential redelivery in case of exception.
• "sessionTransacted" set to "true": transaction confirmation after successful execution of the listener; guaranteed delivery in case of exception.

So, the key to my solution is listenerContainer.setSessionTransacted(true);

Another issue that I encountered was that the JMS provider keeps re-providing the failed message to the same user who failed while processing the message. I do not know if the JMS specification gives a recipe for what the provider should do in such situations, but listenerContainer.shutdown(); worked for me listenerContainer.shutdown(); to disconnect the failed consumer and allow the provider to resend the message and give a chance to another consumer.