How to pass values between two steps ??

[staleks]

Hi all,

I've started to work with spring-batch recently and I found this framework pretty ok.

Now I'm stuck with one problem and I would like to get some help from more experienced users.

Just in short to describe what I'm trying to achieve:

1. Step 1 - I need to read some data from DB, which will fetch records in DB that should later be processed.

2. Step 2 - fetched data (can be multiple records - List<UsersToMove>) must be passed to 2nd step, so step 2 can do some processing of those data. (e.g. Create a mail to admins)

As I've read documentation just to give you the input where I am now:

Job definition

<job id="archiveProfile" job-repository="jobRepository" >
     <step id="fetchProfiles" next="fetchUserSubscriptions">
          <tasklet allow-start-if-complete="true" transaction-manager="liveDBTransactionManager">
	       <chunk reader="profilesToArchiveItemReader" writer="profilesToArchiveItemWriter" commit-interval="${job.commit.interval}"></chunk>
		<listeners>
		   <listener ref="promotionListener" />
		</listeners>
	  </tasklet>
     </step>
     <step id="fetchUserSubscriptions" next="fetchUserSubscriptionBuffers" >
	<tasklet ref="userSubscriptionsToArchiveTask" allow-start-if-complete="true" transaction-manager="liveDBTransactionManager" />
     </step>
</job>

First step item reader and item writer are given respectively:

@Service("profilesToArchiveItemReader")
public class ProfilesToArchiveItemReader implements ItemReader<UserProfileToMove> {

	private static final Logger log = LoggerFactory.getLogger(ProfilesToArchiveItemReader.class);

	private JdbcTemplate liveJdbcTemplate;

	@Autowired
	public ProfilesToArchiveItemReader(@Qualifier("liveDBJdbcTemplate") JdbcTemplate liveJdbcTemplate) {
		setLiveJdbcTemplate(liveJdbcTemplate);
	}

	@Override
	public UserProfileToMove read() throws Exception, UnexpectedInputException, ParseException,
			NonTransientResourceException {
		log.trace("-+- read() method - ENTER -+-");

		Assert.notNull(liveJdbcTemplate);

		String sql = "SELECT DISTINCT p.id, p.activated, p.bad_mail, p.full_member, u.deleted, u.deleted_date, "
				+ "u.last_login_date, u.blocked FROM profile AS p INNER JOIN USER AS u ON u.profile_id = p.id "
				+ "WHERE (p.activated=FALSE AND u.deleted=TRUE)";

		List<UserProfileToMove> result = liveJdbcTemplate.query(sql,
				ParameterizedBeanPropertyRowMapper.newInstance(UserProfileToMove.class));
		log.debug("-+- getAllUserProfilesToMove() method - profilesToMove size: {} -+-", result.size());

		log.info("-+- UserProfileToMove itemRead() -+-");

		log.trace("-+- read() method - LEAVE -+-");
		return result.get(0);
	}

	/**
	 * @param liveJdbcTemplate the liveJdbcTemplate to set
	 */
	public void setLiveJdbcTemplate(JdbcTemplate liveJdbcTemplate) {
		this.liveJdbcTemplate = liveJdbcTemplate;
	}

@Service("profilesToArchiveItemWriter")
public class SavingItemWriter implements ItemWriter<UserProfileToMove> {

	private static final Logger log = LoggerFactory.getLogger(SavingItemWriter.class);

	private StepExecution stepExecution;

	@Override
	public void write(List<? extends UserProfileToMove> items) throws Exception {
		log.trace("-+- write() method - ENTER -+-");

		log.info("-+- profilesToArchiveItemWriter write() -+-");

		ExecutionContext stepContext = stepExecution.getExecutionContext();

		log.info("-+- Items size: {}", items.size());
		stepContext.put("profilesToArchive", items);

		log.trace("-+- write() method - LEAVE -+-");
	}

	@BeforeStep
	public void saveStepExecution(StepExecution stepExecution) {
		this.stepExecution = stepExecution;
	}

So from logs I am pretty sure that everything is fine.

But now I want to have 2nd step as POJO (org.springframework.batch.core.step.tasklet.Metho dInvokingTaskletAdapter)

so this bean I've defined as following:

	<!-- ### Payment package to be moved ### -->		
	<beans:bean id="userSubscriptionsToArchiveTask" class="org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter"		
		p:targetObject-ref="userSubscriptionToMoveBO"
		p:targetMethod="userSubscriptionsToMove" />

Ok, now the idea is to already fetched Users from step ONE, to pass that list
List<UsersToMove> to step TWO so this POJO business service will do some processing on that list.

I've tried to create method like in documentation

 @BeforeStep
    public void retrieveInterstepData(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        ExecutionContext jobContext = jobExecution.getExecutionContext();
        this.someObject = jobContext.get("someKey");
    }

and make it part of Service method, but it never executes, even POJO method <i>userSubscriptionsToMove</i> is called.


Can someone help me or guide me how to accomplish this ?

[staleks]

Now finally get it working Don't know if this is the best way but it works for me.


Solution was pretty simple. I shouldn't use MethodInvokingTaskletAdapter that will trigger POJO, but rather simplier solution:

Just create new tasklet that implements Tasklet interface, and override execute() method.

So this finally works:

	<job id="archiveProfile" job-repository="jobRepository" >
		<step id="fetchProfiles" next="fetchUserSubscriptions">
			<tasklet allow-start-if-complete="true" transaction-manager="liveDBTransactionManager">
				<chunk reader="profilesToArchiveItemReader" writer="profilesToArchiveItemWriter" commit-interval="${job.commit.interval}"></chunk>
				<listeners>
					<listener ref="promotionListener" />
				</listeners>
			</tasklet>
		</step>
		<step id="fetchUserSubscriptions">
			<tasklet ref="userSubscriptionsToArchiveTasklet" allow-start-if-complete="true" transaction-manager="liveDBTransactionManager" />
		</step>
	</job>

and Second step tasklet is given ass following:

@Service("userSubscriptionsToArchiveTasklet")
public class UserSubscriptionsToArchiveTasklet implements Tasklet, InitializingBean {

	private static final Logger log = LoggerFactory.getLogger(UserSubscriptionsToArchiveTasklet.class);

	private static final String JOB_CONTEXT_OBJECT_NAME = "profilesToArchive";

	private List<UserProfileToMove> userProfilesToMove;

	@SuppressWarnings("unchecked")
	@Override
	public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
		log.trace("-+- execute() method - ENTER -+-");

		userProfilesToMove = (List<UserProfileToMove>) GetObjectOutOfExecutionContext.getObjectOutOfExecutionContext(
				chunkContext, JOB_CONTEXT_OBJECT_NAME);
		log.debug("-+- dataToMove : {} -+-", userProfilesToMove.size());

		log.trace("-+- execute() method - LEAVE -+-");
		return RepeatStatus.FINISHED;
	}

     [omit for clarity]

and finally since maybe I am going to make more steps after step 2 that should rely on same data fetched in step one:

public class GetObjectOutOfExecutionContext {

	private static final Logger log = LoggerFactory.getLogger(GetObjectOutOfExecutionContext.class);

	/**
	 * Retreives object that is set in JobContext, to move same data between steps
	 * 
	 * @param chunkContext
	 * @param keyName
	 * @return
	 */
	public static Object getObjectOutOfExecutionContext(ChunkContext chunkContext, String keyName) {
		log.trace("-+- getObjectOutOfExecutionContext() method - ENTER -+-");

		StepContext stepContext = chunkContext.getStepContext();
		StepExecution stepExecution = stepContext.getStepExecution();
		JobExecution jobExecution = stepExecution.getJobExecution();
		ExecutionContext jobContext = jobExecution.getExecutionContext();

		log.trace("-+- getObjectOutOfExecutionContext() method - LEAVE -+-");

		return jobContext.get(keyName);
	}

}

Thank you for reading this, and if anyone has comments please don't hesitate to leave one.

[hyjshanghai]

So instead of "stepContext.put("profilesToArchive", items);", you are now saving the items in the job context, rather than in the step context, right?

[staleks]

Well I didn't find any examples how to save to StepExecutionContext but just find solution for JobExecutionContext.

see following link:
http://static.springsource.org/sprin...aToFutureSteps