Putting Retry in BatchMessageListenerContainer

[gupabhi]

Hi,
I'm want to layer in a RetryTemplate in the BatchMessageListenerContainer.
What I want to do is have the message-processing (listener invocation) called in a RetryTemplate.

My use case is basically to have retry for messages within a Batch.

Tx
{
Repeat (Batch = 10)
{
Retry(Max retry=3)
{
processMsg (invoke listener)
}
}
}

Question: I cant get my head around where exactly the retryTemplate should be layered in. Any help will be appreciated.

Thanks,
Abhi

[dkaminsky]

There are two ways to do this - one is to define your own RepeatOperations, which is a bit complicated and messy. The better way in my opinion is to use AOP. Create a pointcut that includes the methods you want to retry. Then, create an instance of the class:

org.springframework.batch.retry.interceptor.RetryO perationsInterceptor

The advised code will be wrapped in a RetryTemplate as configured through your interceptor. For more information on AOP using Spring, check out: http://static.springframework.org/sp...rence/aop.html

[Dave Syer]

BatchMessageListenerContainer is definitely work in progress - hence it is in the integration test module, not the main release - but I certainly appreciate your interest.

The integration tests exercise the scenario you are interested in (I assume) where the retry forces a rollback and the message is re-presented by the middleware. The most relevant test is in ExternalRetryInBatchTests. But that doesn't use the BatchMessageListenerContainer - it just shows where the retry needs to be applied.

What you need is a stateful retry policy and a retry template wrapping your message processing. You could do that with the ItemReaderRetryPolicy and your own container (as in the integration test), or you could write your own stateful retry policy and wrap just the processing and use the draft BatchMessageListenerContainer, subclassing to add the retry. Maybe you could even re-use the ItemReaderRetryPolicy there as well.

[gupabhi]

Thanks!
Its beginning to make sense to me.

Instead of even sub-classing BatchMessageListenerContainer, I was wondering if I just wrap the listener (passed through container.setMessageListener(...)) call with the retryTemplate.

Something like

public void onMessage(final Message message_, final Session session_) throws JMSException
{
retryTemplate.execute( new ItemProviderRetryCallback( new ItemReaderRecoverer() {
public Object next() throws Exception
{
return message_;
}
}, new AbstractItemWriter() {
public void write(final Object item) {
processMsg(); //do the main processing
}
});
}

Do you see any gotchas in this or do you suggest a more elegant way of achieving this?


Thanks,
Abhi

P.S. Apologies for the bad indentation.

[Dave Syer]

Do you see any gotchas in this or do you suggest a more elegant way of achieving this?

That looks perfectly sane. Good luck.

But you might want to upgrade to the latest release - it hasn't been called ItemProviderRetryCallback for quite some time, and the ItemRecoverer was split off as a separate interface too.

P.S. Apologies for the bad indentation.

Use [code][/code] tags to post code and stack traces.

[gupabhi]

Hi,
I'm facing a problem with using BatchMessageListenerContainer.
When an exception occurs in the the onMessage() call; it goes outside the repeat template but is not caught inside a loop in which the receieveAndExecute() can happen again.
In the following code (from DefaultMessageListenerContainer$AsyncMessageListen erInvoker.run), shouldn't there be a try/catch inside the while-loop surrounding the invokeListener() ? A similar thing happens in ExternalRetryInBatchTests.

					while (isActive()) {
						waitWhileNotRunning();
						if (isActive()) {
							messageReceived = invokeListener();
						}
					}

When an exception occurs in my "processor" (onMessage()) the control just comes out and no retry occurs.

Following is the stacktrace of the exception:

	at mypkg.transport.jms.JMSServer$JmsListener.onMessage(JMSServer.java:246)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:506)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:463)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:435)
	at org.springframework.batch.container.jms.BatchMessageListenerContainer.doExecuteListener(BatchMessageListenerContainer.java:103)
	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:316)
	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255)
	at org.springframework.batch.container.jms.BatchMessageListenerContainer.doBatchCallBack(BatchMessageListenerContainer.java:197)
	at org.springframework.batch.container.jms.BatchMessageListenerContainer$1.doInIteration(BatchMessageListenerContainer.java:157)
	at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:324)
	at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:211)
	at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:142)
	at org.springframework.batch.container.jms.BatchMessageListenerContainer.receiveAndExecute(BatchMessageListenerContainer.java:155)
	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:927)
	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:851)
	at java.lang.Thread.run(Thread.java:595)
Caused by: java.lang.RuntimeException: MsgCallback
	at mypkg.transport.jms.JMSServerTest$3.messageCallback(JMSServerTest.java:100)
	at mypkg.transport.jms.JMSServer$JMSAbstractItemWriter.write(JMSServer.java:287)
	at org.springframework.batch.retry.callback.ItemReaderRetryCallback.process(ItemReaderRetryCallback.java:127)
	at org.springframework.batch.retry.callback.ItemReaderRetryCallback.doWithRetry(ItemReaderRetryCallback.java:100)
	at org.springframework.batch.retry.support.RetryTemplate.execute(RetryTemplate.java:168)
	at mypkg.transport.jms.JMSServer$JmsListener.onMessage(JMSServer.java:244)

Thanks,
Abhi

[Dave Syer]

It's supposed to throw the exception and force a rollback - isn't that what you see? Then the retry happens when the message comes back in in a subsequent transaction.

[gupabhi]

Hi,
Firstly, thanks for you prompt replies. You are correct. There was a problem with my understanding earlier.

-Abhi

[gupabhi]

Hi,
I'm facing a problem with batching in the BatchMessageListenerContainer. I pass to it a repeatTemplate with

            rt.setCompletionPolicy(new SimpleCompletionPolicy(5));

But while processing the messages, it does so one per transaction rather than
processing 5 messages in the same transaction as per completion policy.

On some debugging, I noticed that doBatchCallBack() calls the super.receiveAndExecute() which is in AbstractPollingMessageListenerContainer

In this method, the transaction is explicitly being committed every time, if executed successfully. There is no hook that can be used to tell it to do a commit/rollback conditionally.

Am I missing something here or is this an issue? Please help!!

Thanks,
Abhi