Restart/Continue where left off

[lucasward]

Are you using this class in a multi-threaded step? If so, how have you configured it?

[lpezet]

Are you using this class in a multi-threaded step? If so, how have you configured it?

Nope. Just running it from command line.
There is really nothing more from what I have posted already.

java -Xmx512M -classpath "etc;target/springbatch-test-0.0.1-SNAPSHOT-jar-with-dependencies.jar" org.springframework.batch.core.launch.support.CommandLineJobRunner test_spring_jobs.xml TestSimpleJob uid=123

Each "run" I was mentioning in previous posts are a single run of that previous command.

[lucasward]

How it's launched isn't what I was referring to. You're using atomic integer, etc. So I was assuming you've configured your step to have more than one thread.

[deckingraj]

Hi Lucasward

I dont think i did a good job in explaining my problem. Lemme try once again.

Here is the scenario

Run 1
- Job Failed(due to the forced exception)
- A row is added in the JOB_EXECUTION_TABLE

JOB_EXECUTION_ID=0, VERSION=1, JOB_INSTANCE_ID=0, CREATE_TIME=2008-11-17 10:04:07.318, START_TIME=2008-11-17 10:04:07.365, END_TIME=2008-11-17 10:04:08.02, STATUS=FAILED, CONTINUABLE=N, EXIT_CODE=FAILED

Run 2
- A new Job Passed
- Old Job still lying in Failed state
- Only one new row added in JOB_EXECUTION_TABLE

JOB_EXECUTION_ID=0, VERSION=3, JOB_INSTANCE_ID=0, CREATE_TIME=2008-11-17 10:04:07.318, START_TIME=2008-11-17 10:04:07.365, END_TIME=2008-11-17 10:04:08.02, STATUS=FAILED, CONTINUABLE=N, EXIT_CODE=FAILED

JOB_EXECUTION_ID=1, VERSION=3, JOB_INSTANCE_ID=1, CREATE_TIME=2008-11-17 10:09:03.386, START_TIME=2008-11-17 10:09:03.386, END_TIME=2008-11-17 10:09:06.255, STATUS=COMPLETED, CONTINUABLE=N, EXIT_CODE=COMPLETED


I understand when you say that the failed job might always fail if i have forced the exception before the writer last commit but i am atleast expecting a new row in the Job_Exection table with the same Job_instance(0 in this case) where status might still be "Failed". This will proove that Spring batch tried to invoke the failed job but it failed again due to bad data or whatever reason which unfortunately is not happening in my case

I hope i am clear this time.

[lpezet]

How it's launched isn't what I was referring to. You're using atomic integer, etc. So I was assuming you've configured your step to have more than one thread.

Oh good catch!
So no, I could use a simple int. No threads specified/used/spawn anywhere.
I posted the command line cause it just calls the job defined in the spring config I posted a little before...using the 2 ItemReader+ItemWriter I also posted before, which none mentioned anything about threads, besides the suspicious use of AtomicInteger indeed

[lucasward]

Wait, you're using 2 item readers?

[lpezet]

Wait, you're using 2 item readers?

No I'm not.

[lpezet]

I'll dig into the source code now, but just wanted to post everything in one post in case some gentle soul would like to try to replicate the problem on his/her computer (or might find the answer at a glance )

Version: 1.1.3-RELEASE-A
My ItemReader called TestItemReader:

package test.springbatch;

import java.util.ArrayList;
import java.util.List;

import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.batch.item.NoWorkFoundException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.ResetFailedException;
import org.springframework.batch.item.UnexpectedInputException;

public class TestItemReader implements ItemStream, ItemReader {
	
	private static final List<Integer> NUMBERS = new ArrayList<Integer>();
	private int mIncrementer = 0;
	private int mCurrentItemCount;	
	static {
		for (int i = 0; i < 100; i++) NUMBERS.add(i);
	}
	
	public void open(ExecutionContext pContext) throws ItemStreamException {
		if (pContext.containsKey("read.count")) {
			int oItemCount = Long.valueOf(pContext.getLong("read.count")).intValue();
			try {
				jumpToItem(oItemCount);
			} catch (Exception e) {
				throw new ItemStreamException("Could not move to stored position on restart", e);
			}
			mCurrentItemCount = oItemCount;
		}
	}
	
	private void jumpToItem(int pItemCount) {
		while (mIncrementer++ < pItemCount) {}
	}

	public Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
		if (mIncrementer >= NUMBERS.size()) return null;
		mCurrentItemCount++;		
		return NUMBERS.get(mIncrementer++);
	}	
	
	public void update(ExecutionContext pContext) throws ItemStreamException {
		pContext.putLong("read.count", mCurrentItemCount);
	}
	
	public void close(ExecutionContext pContext) throws ItemStreamException {}	
	public void mark() throws MarkFailedException {}
	public void reset() throws ResetFailedException {}
	
}

My ItemWriter called TestItemWriter:

package test.springbatch;

import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemWriter;

public class TestItemWriter implements ItemWriter {
	
	private int mCounter = 1;
	public void clear() throws ClearFailedException {}
	public void flush() throws FlushFailedException {}

	public void write(Object pObj) throws Exception {
		System.out.println("Writing " + pObj);
		if (mCounter++ >= 10) throw new Exception("It's alright.");
	}
}

My Spring config spring.xml:

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
<beans>
	<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
		<property name="driverClassName" value="com.mysql.jdbc.Driver" />
		<property name="url" value="jdbc:mysql://localhost:3306/testjobs?zeroDateTimeBehavior=convertToNull" />
		<property name="username" value="test" />
		<property name="password" value="test" />
	</bean>
	<bean id="txManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
	  <property name="dataSource" ref="dataSource"/>
	</bean>
	<bean id="jobRepository" class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
	    <property name="databaseType" value="mysql" /> 
	    <property name="dataSource" ref="dataSource" />
	    <property name="transactionManager" ref="txManager" />
	</bean>
	<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
		<property name="jobRepository" ref="jobRepository" />
	</bean>
	<bean id="TestSimpleJob" class="org.springframework.batch.core.job.SimpleJob">
		<property name="jobRepository" ref="jobRepository" />
		<property name="restartable" value="true"/>
		<property name="name" value="TestJob"/>
		<property name="steps">
			<list>
				<bean class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
					<property name="jobRepository" ref="jobRepository" />
					<property name="transactionManager" ref="txManager" />
					<property name="itemReader"><bean class="test.springbatch.TestItemReader"/></property>					
					<property name="itemWriter"><bean class="test.springbatch.TestItemWriter"/></property>
				</bean>				
			</list>
		</property>
	</bean>
</beans>

The command line I run:

java -Xmx512M -classpath "etc;target/springbatch-test-0.0.1-SNAPSHOT-jar-with-dependencies.jar" org.springframework.batch.core.launch.support.CommandLineJobRunner spring.xml TestSimpleJob uid=1

(and yes I'm using Maven...)

Expected output:

> java .....
Writing 0
Writing 1
Writing 2
Writing 3
Writing 4
Writing 5
Writing 6
Writing 7
Writing 8
Writing 9
> java ....
Writing 9
Writing 10
Writing 11
Writing 12
Writing 13
Writing 14
Writing 15
Writing 16
Writing 17
Writing 18
> java ....
Writing 18
Writing 19
Writing 20
Writing 21
Writing 22
...

(or at least something similar where number seems to keep increasing).

I did the same test without the spring config:

        @Test
	public void runJobsWithFailures() throws Exception {
		JobRepository oJobRepo = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao());
		PlatformTransactionManager oTransactionManager = new ResourcelessTransactionManager();
		JobParametersConverter oParamsConverter = new DefaultJobParametersConverter();
		Job oJob = createFullJob(oJobRepo, oTransactionManager);
		SimpleJobLauncher oLauncher = new SimpleJobLauncher();
		oLauncher.setJobRepository(oJobRepo);
		oLauncher.afterPropertiesSet();
		Map<String, String> oParams = new HashMap<String, String>();
		oParams.put("uid", "123");
		JobParameters oJobParams = oParamsConverter.getJobParameters(StringUtils.splitArrayElementsIntoProperties(new String[] { "uid=123"} , "="));
		JobExecution oExecution = null;
                // Run job
		try {
			 oExecution = oLauncher.run(oJob, oJobParams);
		} catch (RepeatException ignore) {}
		// Re-run job
		System.out.println("#1: Re-running job (without using same class instance)");
		oJob = createFullJob(oJobRepo, oTransactionManager);
		try {
			 oExecution = oLauncher.run(oJob, oJobParams);
		} catch (RepeatException ignore) {}
		
		// Re-run job
		System.out.println("#2: Re-running job (without using same class instance)");
		oJob = createFullJob(oJobRepo, oTransactionManager);
		try {
			 oExecution = oLauncher.run(oJob, oJobParams);
		} catch (RepeatException ignore) {}
	}
	
	private Job createFullJob(JobRepository pJobRepo, PlatformTransactionManager pTxManager) throws Exception {
		SimpleJob oJob = new SimpleJob();
		oJob.setName("TestJob");
		oJob.setRestartable(true);		
		oJob.setJobRepository(pJobRepo);
		oJob.afterPropertiesSet();
		Step oStep = createStep(pJobRepo, pTxManager); 		
		oJob.addStep(oStep);
		return oJob;
	}
	
	private Step createStep(JobRepository pJobRepo, PlatformTransactionManager pTxManager) throws Exception {
		SimpleStepFactoryBean oStepFactory = new SimpleStepFactoryBean();				
		oStepFactory.setItemReader(new TestItemReader());
		oStepFactory.setItemWriter(new TestItemWriter());
		oStepFactory.setTransactionManager(pTxManager);
		oStepFactory.setJobRepository(pJobRepo);
		oStepFactory.setBeanName("TestStep");
		return (Step) oStepFactory.getObject();
	}

And it works fine.

It doesn't look that much different from my config in spring.xml but the behavior is different.

Still clueless

[lpezet]

Problem seems to come from something dealing with the database.
I changed the code not using spring.xml and setup the JobRepository to use MySQL DB instead of the In-Memory one.

Output with In-Memoty JobRepository:

Writing 0
Writing 1
Writing 2
Writing 3
Writing 4
Writing 5
Writing 6
Writing 7
Writing 8
Writing 9
#1: Re-running job (without using same class instance)
Writing 10
Writing 11
Writing 12
Writing 13
Writing 14
Writing 15
Writing 16
Writing 17
Writing 18
Writing 19
#2: Re-running job (without using same class instance)
Writing 19
Writing 20
Writing 21
Writing 22
Writing 23
Writing 24
Writing 25
Writing 26
Writing 27
Writing 28

Output with the same exact code but using MySQL DB:

Writing 0
Writing 1
Writing 2
Writing 3
Writing 4
Writing 5
Writing 6
Writing 7
Writing 8
Writing 9
#1: Re-running job (without using same class instance)
Writing 10
Writing 11
Writing 12
Writing 13
Writing 14
Writing 15
Writing 16
Writing 17
Writing 18
Writing 19
#2: Re-running job (without using same class instance)
(Exception)
java.lang.IllegalStateException: There must be at most one latest job execution
	at org.springframework.util.Assert.state(Assert.java:384)
	at org.springframework.batch.core.repository.dao.JdbcJobExecutionDao.getLastJobExecution(JdbcJobExecutionDao.java:237)
	at org.springframework.batch.core.repository.support.SimpleJobRepository.createJobExecution(SimpleJobRepository.java:179)
	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)
	...

Sometimes it would work just fine with MySQL, sometimes it would break right before running the third run and throws the "IllegalStateException" as shown above.

Maybe I should just upgrade to 2.0.0...

[lpezet]

Here is the answer: the name of the step MUST be the same.
It's really a stupid-simple thing I know...sigh
So the following config will not work as expected:

<bean id="TestSimpleJob" class="org.springframework.batch.core.job.SimpleJob">
	<property name="jobRepository" ref="jobRepository" />
	<property name="restartable" value="true"/>
	<property name="name" value="TestJob"/>
	<property name="steps">
		<list>
			<bean class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
				<property name="jobRepository" ref="jobRepository" />
				<property name="transactionManager" ref="txManager" />
				<property name="itemReader"><bean class="test.springbatch.TestItemReader"/></property>					
				<property name="itemWriter"><bean class="test.springbatch.TestItemWriter"/></property>
			</bean>				
		</list>
	</property>
</bean>

The name of the step will be the name of the instance of the class (like you would get doing myObject.toString()).
In the DB, it then looks like this (in the batch_step_execution table):
org.springframework.batch.core.step.item.SimpleSte pFactoryBean#18c56d
So of course, between runs, the instance won't have the same "id" (in this case "18c56d")...but sometimes it does. That's why I was having this weird random behavior.

The fix is trivial then:

<bean id="rwStep" class="org.springframework.batch.core.step.item.SimpleStepFactoryBean">
	<property name="jobRepository" ref="jobRepository" />
	<property name="transactionManager" ref="txManager" />
	<property name="itemReader"><bean class="test.springbatch.TestItemReader"/></property>					
	<property name="itemWriter"><bean class="test.springbatch.TestItemWriter"/></property>
</bean>
<bean id="TestSimpleJob" class="org.springframework.batch.core.job.SimpleJob">
	<property name="jobRepository" ref="jobRepository" />
	<property name="restartable" value="true"/>
	<property name="name" value="TestJob"/>
	<property name="steps">
		<list>
			<ref bean="rwStep"/>		
		</list>
	</property>
</bean>

Now the name of the step is "rwStep" and that's what end up in the db.
If the job fails (e.g. throws an exception) it will resume as expected on next run.