Retry in case of a failed reading

[sebge2]

Hello,

I'm trying resty functionalities of spring batch. Unfortunately I face some problems.

First I define an abstract step:

	<bean id="defaultRetryPolicy" class="com.bsb.sf.batch.DefaultRetryPolicy" />

	<bean id="fatherStep"
		class="org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean" abstract="true">
		<property name="transactionManager" ref="transactionManager" />
		<property name="jobRepository" ref="jobRepository" />
		<property name="backOffPolicy">
			<bean class="org.springframework.batch.retry.backoff.FixedBackOffPolicy">
				<property name="backOffPeriod" value="8000" />
			</bean>
		</property>
		
		<property name="retryableExceptionClasses">
			<map>
			</map>
		</property>
		<property name="retryLimit" value="3" />
		<property name="retryPolicy" ref="defaultRetryPolicy" />
	</bean>


	<!--parent="defaultBatchJob"-->
	<job id="retryPolicyJob" xmlns="http://www.springframework.org/schema/batch">
		<step id="mainStep" parent="fatherStep">
			<tasklet>
				<chunk reader="itemReader" processor="itemProcessor" writer="itemWriter"
					commit-interval="2">					
				</chunk>				
			</tasklet>			
		</step>
	</job>

If an exception is thrown when committing (the first commit), my skip policy is called. This also works when it's processing the first entity. But, if an exception is thrown while reading the first entity, my retry policy (cfr defaultRetryPolicy) is not called. Is it a bug ?

This is my console:

2010-01-11 11:08:17 [main] DataSourceInitializer [DEBUG] Executing script: schema-drop-hsqldb.sql
2010-01-11 11:08:18 [main] DataSourceInitializer [DEBUG] Executing script: batch-model-drop-h2.sql
2010-01-11 11:08:18 [main] DataSourceInitializer [DEBUG] Executing script: schema-drop-hsqldb.sql
2010-01-11 11:08:18 [main] DataSourceInitializer [DEBUG] Executing script: schema-hsqldb.sql
2010-01-11 11:08:18 [main] DataSourceInitializer [DEBUG] Executing script: batch-model-create-h2.sql
2010-01-11 11:08:19 [main] ExceptionItemReader [DEBUG] Exception class: com.bsb.sf.batch.MyRetryableException
2010-01-11 11:08:19 [main] ExceptionItemReader [DEBUG] Account reader nbrun = 0 maxrun: 2
2010-01-11 11:08:19 [main] ExceptionItemReader [DEBUG] Crash, item 1
2010-01-11 11:08:19 [main] AbstractStep [ERROR] Encountered an error executing the step
org.springframework.batch.core.step.skip.NonSkippableReadException: Non-skippable exception during read
	at org.springframework.batch.core.step.item.FaultTolerantChunkProvider.read(FaultTolerantChunkProvider.java:81)
	at org.springframework.batch.core.step.item.SimpleChunkProvider$1.doInIteration(SimpleChunkProvider.java:106)
	at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:367)
	at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:214)
	at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:143)
	at org.springframework.batch.core.step.item.SimpleChunkProvider.provide(SimpleChunkProvider.java:103)
	at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:68)
	at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:347)
	at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:128)
	at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:261)
	at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:76)
	at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:367)
	at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:214)
	at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:143)
	at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:247)
	at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:196)
	at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:115)
	at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:61)
	at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:60)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:144)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:124)
	at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:99)
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:276)
	at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:118)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:49)
	at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:112)
	at com.bsb.sf.incubator.batch.AbstractBatchJobLauncherTest.doLaunchJob(AbstractBatchJobLauncherTest.java:62)
	at com.bsb.sf.batch.DefaultRetryPolicyTest.mikeyWorld(DefaultRetryPolicyTest.java:89)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
	at java.lang.reflect.Method.invoke(Unknown Source)
	at org.springframework.test.context.junit4.SpringTestMethod.invoke(SpringTestMethod.java:160)
	at org.springframework.test.context.junit4.SpringMethodRoadie.runTestMethod(SpringMethodRoadie.java:233)
	at org.springframework.test.context.junit4.SpringMethodRoadie$RunBeforesThenTestThenAfters.run(SpringMethodRoadie.java:333)
	at org.springframework.test.context.junit4.SpringMethodRoadie.runWithRepetitions(SpringMethodRoadie.java:217)
	at org.springframework.test.context.junit4.SpringMethodRoadie.runTest(SpringMethodRoadie.java:197)
	at org.springframework.test.context.junit4.SpringMethodRoadie.run(SpringMethodRoadie.java:143)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.invokeTestMethod(SpringJUnit4ClassRunner.java:160)
	at org.junit.internal.runners.JUnit4ClassRunner.runMethods(JUnit4ClassRunner.java:51)
	at org.junit.internal.runners.JUnit4ClassRunner$1.run(JUnit4ClassRunner.java:44)
	at org.junit.internal.runners.ClassRoadie.runUnprotected(ClassRoadie.java:27)
	at org.junit.internal.runners.ClassRoadie.runProtected(ClassRoadie.java:37)
	at org.junit.internal.runners.JUnit4ClassRunner.run(JUnit4ClassRunner.java:42)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:97)
	at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:46)
	at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197)
Caused by: com.bsb.sf.batch.MyRetryableException: Exception planned to be launch at item 1
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source)
	at java.lang.reflect.Constructor.newInstance(Unknown Source)
	at com.bsb.sf.incubator.batch.simple.ExceptionItemReader.read(ExceptionItemReader.java:78)
	at com.bsb.sf.incubator.batch.simple.ExceptionItemReader.read(ExceptionItemReader.java:1)
	at org.springframework.batch.core.step.item.SimpleChunkProvider.doRead(SimpleChunkProvider.java:90)
	at org.springframework.batch.core.step.item.FaultTolerantChunkProvider.read(FaultTolerantChunkProvider.java:68)
	... 50 more
2010-01-11 11:08:20 [Thread-22] DataSourceInitializer [DEBUG] Executing script: schema-drop-hsqldb.sql
2010-01-11 11:08:20 [Thread-22] DataSourceInitializer [DEBUG] Executing script: batch-model-drop-h2.sql

Best regards,

Sébastien

[Dave Syer]

You haven't defined any skip semantics (only retry). I'm confused about what you think the problem might be.

[sebge2]

I've understood: reading is not retryable i.e., if an exception occurs while reading, the skip policy is immediately called because it's not possible to "retry" a reading (the retry policy is not called in that case)

[dma_k]

I vote for the same feature.

I have found out, that if "retry-policy" attribute is defined for "<batch:chunk>" (no other settings concerning rollback / skip exceptions), the given policy implementation is not involved for item reader (only for item processor / writer). This is a pity.

I was looking into FaultTolerantStepFactoryBean, methods configureChunkProvider() (has no notion about retry, but is expected to) and configureChunkProcessor() (handles retry policy as expected).

[Dave Syer]

On reflection I don't think we should support retry of the ItemReader - the semantics are too hard to define. What is you actual use case here? What does your ItemReader do that is retryable? The problem in general is that ItemReader is a "forward only" contract - it's not possible in general to implement readers that can go back to a previous position in the stream with the interface we have (and we are not going to make it more complicated without a good reason). Given that, if a read() fails, what is the reader supposed to do when we "retry" the operation? Go to the next item is the only possible choice, and that is functionally the same as a skip.

There might be some use cases where you wan to control the calls to read() using a RetryPolicy. In that case you could use a RetryOperationsInterceptor around your ItemReader. But I think the end result would still be equivalent to a skip (with less callbacks to the skip listener).

[dma_k]

On reflection I don't think we should support retry of the ItemReader - the semantics are too hard to define. What is you actual use case here? What does your ItemReader do that is retryable?

I agree, that in most cases the exceptions from reader should be handled specifically (skip record or terminate the processing). But I can imagine at least two cases, when retry is fine for reader:

• In my case I read the items from WebService. When the service is overloaded, it raises the specific RemoteAccessException with message "Too busy. Please try again later." The reader I am using is stateful and maintains the current "cursor position" for the current WebService request, which is incremented when the WebService request is succeeded:
  

  myService.getItem(rangeOffset); // exception can be raised
  rangeOffset++; // read() can be safely called by framework again, as the counter was not incremented in case of exception

• Another case which I can easily be faced against is the table row locking. Imagine I have a job A that inserts the rows into the table and another job B reading them. Depending on the setting of DB, job B may fail if job A has inserted the row and not yet committed. DB2 raises locking exception, which I can't control (databases are maintained by independent team). In this case I think nice to have ItemStream.open() retryable.

The problem in general is that ItemReader is a "forward only" contract - it's not possible in general to implement readers that can go back to a previous position in the stream with the interface we have (and we are not going to make it more complicated without a good reason). Given that, if a read() fails, what is the reader supposed to do when we "retry" the operation? Go to the next item is the only possible choice, and that is functionally the same as a skip.

There might be some use cases where you wan to control the calls to read() using a RetryPolicy. In that case you could use a RetryOperationsInterceptor around your ItemReader. But I think the end result would still be equivalent to a skip (with less callbacks to the skip listener).

Maybe you are right: in this context retry = skip for the framework, as it basically have to call read() again. However "skip count" does not help me to do a correct retry (for example, if I want to increase the retry interval between attempts), so the implementer should be a ChunkListener to reset the retry count.

So what would be your final recommendation:

1. Use AOP advising via RetryOperationsInterceptor + RetryTemplate + RetryPolicy
2. Implement SkipPolicy and use some techniques to reset the "retry counter"
3. Something else

[Dave Syer]

The use cases are helpful, but neither has changed my mind. As long as you only increment your cursor counter when the webservice call is successful there is no problem with the first one, as far as state management goes anyway. And the database read in the second example is not in principle any different (although I can see in Spring Batch that we need to re-order the code in the JDBC readers).

My recommendation is:

* If you care about skip counts and don't want a failed read to show up as a skip use AOP
* Otherwise declare the exception as skippable
* Raise a JIRA for the JDBC reader corner case so we only increment the counter when the read succeeds

[dma_k]

As long as you only increment your cursor counter when the webservice call is successful there is no problem with the first one, as far as state management goes anyway. And the database read in the second example is not in principle any different (although I can see in Spring Batch that we need to re-order the code in the JDBC readers).

Dave, thank you for comments. I am afraid, I have not enough information to open a JIRA issue for JDBC part, as I don't have a clear test case and I simply cannot verify the ticket & close it. If you fee you can make a code better with some minimalistic change – you don't need a ticket for that. Thanks for the note conc statistics: indeed the increasing skip counter may confuse somebody later.

I will go for AOP (even though it adds one more dependency to the project), as this approach keeps the separation of responsibilities clear.